Merge origin/main into fix-ollama-image-generation

tests/tools/test_apply_patch_tool.py

@@ -0,0 +1,330 @@
+from __future__ import annotations
+
+import asyncio
+
+from nanobot.agent.tools.apply_patch import ApplyPatchTool
+
+
+def test_apply_patch_edits_replace(tmp_path):
+    target = tmp_path / "calc.py"
+    target.write_text("def add(a, b):\n    return a + b\n")
+    tool = ApplyPatchTool(workspace=tmp_path)
+
+    result = asyncio.run(
+        tool.execute(
+            edits=[
+                {
+                    "path": "calc.py",
+                    "action": "replace",
+                    "old_text": "    return a + b",
+                    "new_text": "    return a - b",
+                }
+            ]
+        )
+    )
+
+    assert "update calc.py" in result
+    assert target.read_text() == "def add(a, b):\n    return a - b\n"
+
+
+def test_apply_patch_edits_add_new_file(tmp_path):
+    tool = ApplyPatchTool(workspace=tmp_path)
+
+    result = asyncio.run(
+        tool.execute(
+            edits=[
+                {
+                    "path": "config.py",
+                    "action": "add",
+                    "new_text": "DEBUG = True",
+                }
+            ]
+        )
+    )
+
+    assert "add config.py" in result
+    assert (tmp_path / "config.py").read_text() == "DEBUG = True\n"
+
+
+def test_apply_patch_edits_preserves_new_file_trailing_blank_lines(tmp_path):
+    tool = ApplyPatchTool(workspace=tmp_path)
+
+    result = asyncio.run(
+        tool.execute(
+            edits=[
+                {
+                    "path": "notes.txt",
+                    "action": "add",
+                    "new_text": "one\n\n",
+                }
+            ]
+        )
+    )
+
+    assert "add notes.txt" in result
+    assert (tmp_path / "notes.txt").read_text() == "one\n\n"
+
+
+def test_apply_patch_edits_add_to_existing_file(tmp_path):
+    target = tmp_path / "log.py"
+    target.write_text("import logging\n\nlogger = logging.getLogger(__name__)\n")
+    tool = ApplyPatchTool(workspace=tmp_path)
+
+    result = asyncio.run(
+        tool.execute(
+            edits=[
+                {
+                    "path": "log.py",
+                    "action": "add",
+                    "new_text": "def debug(msg):\n    logger.debug(msg)",
+                }
+            ]
+        )
+    )
+
+    assert "update log.py" in result
+    assert (
+        target.read_text()
+        == "import logging\n\nlogger = logging.getLogger(__name__)\ndef debug(msg):\n    logger.debug(msg)\n"
+    )
+
+
+def test_apply_patch_edits_delete(tmp_path):
+    target = tmp_path / "utils.py"
+    target.write_text("def unused():\n    pass\ndef used():\n    return 1\n")
+    tool = ApplyPatchTool(workspace=tmp_path)
+
+    result = asyncio.run(
+        tool.execute(
+            edits=[
+                {
+                    "path": "utils.py",
+                    "action": "delete",
+                    "old_text": "def unused():\n    pass\n",
+                }
+            ]
+        )
+    )
+
+    assert "update utils.py" in result
+    assert target.read_text() == "def used():\n    return 1\n"
+
+
+def test_apply_patch_edits_delete_entire_file(tmp_path):
+    target = tmp_path / "obsolete.txt"
+    target.write_text("remove me\n")
+    tool = ApplyPatchTool(workspace=tmp_path)
+
+    result = asyncio.run(
+        tool.execute(
+            edits=[
+                {
+                    "path": "obsolete.txt",
+                    "action": "delete",
+                    "old_text": "remove me\n",
+                }
+            ]
+        )
+    )
+
+    assert "delete obsolete.txt" in result
+    assert not target.exists()
+
+
+def test_apply_patch_edits_delete_substring_with_surrounding_whitespace(tmp_path):
+    target = tmp_path / "keep_whitespace.txt"
+    target.write_text("  token  \n")
+    tool = ApplyPatchTool(workspace=tmp_path)
+
+    result = asyncio.run(
+        tool.execute(
+            edits=[
+                {
+                    "path": "keep_whitespace.txt",
+                    "action": "delete",
+                    "old_text": "token",
+                }
+            ]
+        )
+    )
+
+    assert "update keep_whitespace.txt" in result
+    assert target.exists()
+    assert target.read_text() == "    \n"
+
+
+def test_apply_patch_edits_batch_multiple_files(tmp_path):
+    a = tmp_path / "a.py"
+    a.write_text("X = 1\n")
+    b = tmp_path / "b.py"
+    b.write_text("from a import X\nprint(X)\n")
+    tool = ApplyPatchTool(workspace=tmp_path)
+
+    result = asyncio.run(
+        tool.execute(
+            edits=[
+                {
+                    "path": "a.py",
+                    "action": "replace",
+                    "old_text": "X = 1",
+                    "new_text": "Y = 1",
+                },
+                {
+                    "path": "b.py",
+                    "action": "replace",
+                    "old_text": "from a import X",
+                    "new_text": "from a import Y",
+                },
+            ]
+        )
+    )
+
+    assert "update a.py" in result
+    assert "update b.py" in result
+    assert a.read_text() == "Y = 1\n"
+    assert b.read_text() == "from a import Y\nprint(X)\n"
+
+
+def test_apply_patch_edits_rejects_ambiguous_old_text(tmp_path):
+    target = tmp_path / "repeated.txt"
+    target.write_text("target\nmiddle\ntarget\n")
+    tool = ApplyPatchTool(workspace=tmp_path)
+
+    result = asyncio.run(
+        tool.execute(
+            edits=[
+                {
+                    "path": "repeated.txt",
+                    "action": "replace",
+                    "old_text": "target",
+                    "new_text": "changed",
+                }
+            ]
+        )
+    )
+
+    assert "old_text appears multiple times" in result
+    assert target.read_text() == "target\nmiddle\ntarget\n"
+
+
+def test_apply_patch_edits_dry_run_validates_without_writing(tmp_path):
+    target = tmp_path / "dry.txt"
+    target.write_text("before\n")
+    tool = ApplyPatchTool(workspace=tmp_path)
+
+    result = asyncio.run(
+        tool.execute(
+            edits=[
+                {
+                    "path": "dry.txt",
+                    "action": "replace",
+                    "old_text": "before",
+                    "new_text": "after",
+                },
+                {
+                    "path": "added.txt",
+                    "action": "add",
+                    "new_text": "new",
+                },
+            ],
+            dry_run=True,
+        )
+    )
+
+    assert "Patch dry-run succeeded" in result
+    assert target.read_text() == "before\n"
+    assert not (tmp_path / "added.txt").exists()
+
+
+def test_apply_patch_edits_rejects_absolute_and_parent_paths(tmp_path):
+    tool = ApplyPatchTool(workspace=tmp_path)
+
+    absolute = asyncio.run(
+        tool.execute(
+            edits=[
+                {
+                    "path": "/tmp/owned.txt",
+                    "action": "add",
+                    "new_text": "nope",
+                }
+            ]
+        )
+    )
+    parent = asyncio.run(
+        tool.execute(
+            edits=[
+                {
+                    "path": "../owned.txt",
+                    "action": "add",
+                    "new_text": "nope",
+                }
+            ]
+        )
+    )
+    windows_absolute = asyncio.run(
+        tool.execute(
+            edits=[
+                {
+                    "path": r"C:\owned.txt",
+                    "action": "add",
+                    "new_text": "nope",
+                }
+            ]
+        )
+    )
+    windows_parent = asyncio.run(
+        tool.execute(
+            edits=[
+                {
+                    "path": r"..\owned.txt",
+                    "action": "add",
+                    "new_text": "nope",
+                }
+            ]
+        )
+    )
+
+    assert "must be relative" in absolute
+    assert "must not contain '..'" in parent
+    assert "must be relative" in windows_absolute
+    assert "must not contain '..'" in windows_parent
+    assert not (tmp_path.parent / "owned.txt").exists()
+
+
+def test_apply_patch_edits_reports_invalid_edit_shapes(tmp_path):
+    tool = ApplyPatchTool(workspace=tmp_path)
+
+    missing_path = asyncio.run(tool.execute(edits=[{"action": "add", "new_text": "x"}]))
+    missing_action = asyncio.run(tool.execute(edits=[{"path": "x.txt", "new_text": "x"}]))
+    non_object = asyncio.run(tool.execute(edits=["not an object"]))  # type: ignore[list-item]
+
+    assert "path required for edit" in missing_path
+    assert "action required for edit: x.txt" in missing_action
+    assert "each edit must be an object" in non_object
+
+
+def test_apply_patch_edits_rolls_back_when_late_operation_fails(tmp_path):
+    first = tmp_path / "first.txt"
+    first.write_text("before\n")
+    tool = ApplyPatchTool(workspace=tmp_path)
+
+    result = asyncio.run(
+        tool.execute(
+            edits=[
+                {
+                    "path": "first.txt",
+                    "action": "replace",
+                    "old_text": "before",
+                    "new_text": "after",
+                },
+                {
+                    "path": "missing.txt",
+                    "action": "delete",
+                    "old_text": "remove me",
+                },
+            ]
+        )
+    )
+
+    assert "file to update does not exist: missing.txt" in result
+    assert first.read_text() == "before\n"

tests/tools/test_edit_enhancements.py

@@ -1,5 +1,5 @@
 """Tests for EditFileTool enhancements: read-before-edit tracking, path suggestions,
-.ipynb detection, and create-file semantics."""
+notebook JSON editing, and create-file semantics."""
 
 import pytest
 
@@ -108,22 +108,27 @@ class TestEditCreateFile:
 
 
 # ---------------------------------------------------------------------------
-# .ipynb detection
+# .ipynb editing
 # ---------------------------------------------------------------------------
 
-class TestEditIpynbDetection:
-    """edit_file should refuse .ipynb and suggest notebook_edit."""
+class TestEditIpynbFiles:
+    """edit_file edits notebooks as normal JSON files."""
 
     @pytest.fixture()
     def tool(self, tmp_path):
         return EditFileTool(workspace=tmp_path)
 
     @pytest.mark.asyncio
-    async def test_ipynb_rejected_with_suggestion(self, tool, tmp_path):
+    async def test_ipynb_can_be_edited_as_json(self, tool, tmp_path):
         f = tmp_path / "analysis.ipynb"
         f.write_text('{"cells": []}', encoding="utf-8")
-        result = await tool.execute(path=str(f), old_text="x", new_text="y")
-        assert "notebook" in result.lower()
+        result = await tool.execute(
+            path=str(f),
+            old_text='"cells": []',
+            new_text='"cells": [{"cell_type": "markdown", "source": "hi"}]',
+        )
+        assert "Successfully edited" in result
+        assert '"source": "hi"' in f.read_text(encoding="utf-8")
 
 
 # ---------------------------------------------------------------------------

tests/tools/test_exec_platform.py

@@ -162,7 +162,7 @@ class TestPathAppendPlatform:
         captured_cmd = None
         captured_env = {}
 
-        async def capture_spawn(cmd, cwd, env):
+        async def capture_spawn(cmd, cwd, env, shell_program=None, login=True):
             nonlocal captured_cmd
             captured_cmd = cmd
             captured_env.update(env)
@@ -190,7 +190,7 @@ class TestPathAppendPlatform:
 
         captured_env = {}
 
-        async def capture_spawn(cmd, cwd, env):
+        async def capture_spawn(cmd, cwd, env, shell_program=None, login=True):
             captured_env.update(env)
             return mock_proc
 

tests/tools/test_exec_session_tools.py

@@ -0,0 +1,361 @@
+from __future__ import annotations
+
+import asyncio
+import re
+import shlex
+import subprocess
+import sys
+
+from nanobot.agent.tools.shell import ExecTool
+from nanobot.agent.tools.exec_session import ExecSessionManager, ListExecSessionsTool, WriteStdinTool
+
+
+def _python_command(code: str) -> str:
+    if sys.platform == "win32":
+        return f"{subprocess.list2cmdline([sys.executable])} -u -c {subprocess.list2cmdline([code])}"
+    return f"{shlex.quote(sys.executable)} -u -c {shlex.quote(code)}"
+
+
+def _session_id(output: str) -> str:
+    match = re.search(r"session_id:\s*([0-9a-f]+)", output)
+    assert match, output
+    return match.group(1)
+
+
+def test_exec_keeps_one_shot_behavior_without_yield_time_ms(tmp_path):
+    async def run() -> str:
+        tool = ExecTool(working_dir=str(tmp_path), timeout=5)
+        return await tool.execute(command="echo hello")
+
+    result = asyncio.run(run())
+
+    assert "hello" in result
+    assert "Exit code: 0" in result
+    assert "session_id:" not in result
+
+
+def test_exec_accepts_command_aliases(tmp_path):
+    async def run() -> str:
+        tool = ExecTool(working_dir="/")
+        return await tool.execute(
+            cmd=_python_command("import os; print(os.getcwd())"),
+            workdir=str(tmp_path),
+        )
+
+    result = asyncio.run(run())
+
+    assert str(tmp_path) in result
+    assert "Exit code: 0" in result
+
+
+def test_exec_returns_completed_session_output_when_yield_time_ms_is_used(tmp_path):
+    async def run() -> str:
+        manager = ExecSessionManager()
+        tool = ExecTool(working_dir=str(tmp_path), timeout=5, session_manager=manager)
+        stdin_tool = WriteStdinTool(manager=manager)
+
+        result = await tool.execute(command="echo hello", yield_time_ms=1000)
+        if "session_id:" in result:
+            sid = _session_id(result)
+            result += "\n" + await stdin_tool.execute(
+                session_id=sid,
+                chars="",
+                yield_time_ms=1000,
+            )
+        return result
+
+    result = asyncio.run(run())
+
+    assert "hello" in result
+    assert "Exit code: 0" in result
+    assert "session_id:" not in result
+
+
+def test_exec_session_accepts_max_output_tokens_alias(tmp_path):
+    async def run() -> str:
+        manager = ExecSessionManager()
+        tool = ExecTool(working_dir=str(tmp_path), timeout=5, session_manager=manager)
+        command = _python_command("print('A' * 2000)")
+        return await tool.execute(
+            command=command,
+            yield_time_ms=1000,
+            max_output_tokens=1000,
+        )
+
+    result = asyncio.run(run())
+
+    assert "chars truncated" in result
+    assert "Exit code: 0" in result
+
+
+def test_exec_one_shot_accepts_max_output_tokens_alias(tmp_path):
+    async def run() -> str:
+        tool = ExecTool(working_dir=str(tmp_path), timeout=5)
+        command = _python_command("print('A' * 2000)")
+        return await tool.execute(command=command, max_output_tokens=1000)
+
+    result = asyncio.run(run())
+
+    assert "chars truncated" in result
+    assert "Exit code: 0" in result
+
+
+def test_exec_accepts_supported_shell_parameter(tmp_path):
+    async def run() -> str:
+        tool = ExecTool(working_dir=str(tmp_path), timeout=5)
+        return await tool.execute(command="echo shell-ok", shell="sh", login=False)
+
+    if sys.platform == "win32":
+        return
+    result = asyncio.run(run())
+
+    assert "shell-ok" in result
+    assert "Exit code: 0" in result
+
+
+def test_exec_rejects_unsupported_shell(tmp_path):
+    async def run() -> str:
+        tool = ExecTool(working_dir=str(tmp_path), timeout=5)
+        return await tool.execute(command="echo no", shell="python")
+
+    if sys.platform == "win32":
+        return
+    result = asyncio.run(run())
+
+    assert "unsupported shell" in result
+
+
+def test_exec_can_continue_with_stdin(tmp_path):
+    async def run() -> tuple[str, str]:
+        manager = ExecSessionManager()
+        exec_tool = ExecTool(working_dir=str(tmp_path), timeout=5, session_manager=manager)
+        stdin_tool = WriteStdinTool(manager=manager)
+        command = _python_command(
+            "import sys; print('ready', flush=True); "
+            "line=sys.stdin.readline(); print('got:' + line.strip(), flush=True)"
+        )
+
+        initial = await exec_tool.execute(command=command, yield_time_ms=500)
+        sid = _session_id(initial)
+        result = await stdin_tool.execute(session_id=sid, chars="ping\n", yield_time_ms=1000)
+        return initial, result
+
+    initial, result = asyncio.run(run())
+    assert "ready" in initial
+    assert "Process running" in initial
+    assert "Elapsed:" in initial
+    assert "got:ping" in result
+    assert "Exit code: 0" in result
+    assert "Elapsed:" in result
+
+
+def test_write_stdin_can_close_stdin(tmp_path):
+    async def run() -> tuple[str, str]:
+        manager = ExecSessionManager()
+        exec_tool = ExecTool(working_dir=str(tmp_path), timeout=5, session_manager=manager)
+        stdin_tool = WriteStdinTool(manager=manager)
+        command = _python_command(
+            "import sys; print('ready', flush=True); "
+            "data=sys.stdin.read(); print('got:' + data, flush=True)"
+        )
+
+        initial = await exec_tool.execute(command=command, yield_time_ms=500)
+        sid = _session_id(initial)
+        result = await stdin_tool.execute(
+            session_id=sid,
+            chars="payload",
+            close_stdin=True,
+            yield_time_ms=1000,
+        )
+        return initial, result
+
+    initial, result = asyncio.run(run())
+    assert "ready" in initial
+    assert "got:payload" in result
+    assert "Stdin closed." in result
+    assert "Exit code: 0" in result
+
+
+def test_write_stdin_can_terminate_session(tmp_path):
+    async def run() -> tuple[str, str]:
+        manager = ExecSessionManager()
+        exec_tool = ExecTool(working_dir=str(tmp_path), timeout=30, session_manager=manager)
+        stdin_tool = WriteStdinTool(manager=manager)
+        command = _python_command(
+            "import time; print('ready', flush=True); time.sleep(30)"
+        )
+
+        initial = await exec_tool.execute(command=command, yield_time_ms=500)
+        sid = _session_id(initial)
+        result = await stdin_tool.execute(
+            session_id=sid,
+            terminate=True,
+            yield_time_ms=0,
+        )
+        return initial, result
+
+    initial, result = asyncio.run(run())
+    assert "ready" in initial
+    assert "Session terminated." in result
+    assert "Exit code:" in result
+
+
+def test_write_stdin_accepts_max_output_tokens_alias(tmp_path):
+    async def run() -> tuple[str, str, str]:
+        manager = ExecSessionManager()
+        exec_tool = ExecTool(working_dir=str(tmp_path), timeout=5, session_manager=manager)
+        stdin_tool = WriteStdinTool(manager=manager)
+        command = _python_command(
+            "import time; print('A' * 2000, flush=True); time.sleep(5)"
+        )
+
+        initial = await exec_tool.execute(command=command, yield_time_ms=0)
+        sid = _session_id(initial)
+        poll = await stdin_tool.execute(
+            session_id=sid,
+            yield_time_ms=500,
+            max_output_tokens=1000,
+        )
+        cleanup = await stdin_tool.execute(session_id=sid, terminate=True, yield_time_ms=0)
+        return initial, poll, cleanup
+
+    initial, poll, cleanup = asyncio.run(run())
+    assert "Process running" in initial
+    assert "chars truncated" in poll
+    assert "Session terminated." in cleanup
+
+
+def test_write_stdin_preserves_completed_session_output_until_polled(tmp_path):
+    async def run() -> tuple[str, str]:
+        manager = ExecSessionManager()
+        exec_tool = ExecTool(working_dir=str(tmp_path), timeout=5, session_manager=manager)
+        stdin_tool = WriteStdinTool(manager=manager)
+        command = _python_command(
+            "import time; print('ready', flush=True); "
+            "time.sleep(1.0); print('done', flush=True)"
+        )
+
+        initial = await exec_tool.execute(command=command, yield_time_ms=300)
+        sid = _session_id(initial)
+        await asyncio.sleep(1.2)
+        final = await stdin_tool.execute(session_id=sid, chars="", yield_time_ms=0)
+        return initial, final
+
+    initial, final = asyncio.run(run())
+
+    assert "ready" in initial
+    assert "done" in final
+    assert "Exit code: 0" in final
+
+
+def test_write_stdin_can_wait_for_expected_output(tmp_path):
+    async def run() -> tuple[str, str, str]:
+        manager = ExecSessionManager()
+        exec_tool = ExecTool(working_dir=str(tmp_path), timeout=5, session_manager=manager)
+        stdin_tool = WriteStdinTool(manager=manager)
+        command = _python_command(
+            "import time; print('booting', flush=True); "
+            "time.sleep(0.4); print('ready', flush=True); time.sleep(5)"
+        )
+
+        initial = await exec_tool.execute(command=command, yield_time_ms=100)
+        sid = _session_id(initial)
+        waited = await stdin_tool.execute(
+            session_id=sid,
+            wait_for="ready",
+            wait_timeout_ms=3000,
+            yield_time_ms=0,
+        )
+        cleanup = await stdin_tool.execute(session_id=sid, terminate=True, yield_time_ms=0)
+        return initial, waited, cleanup
+
+    initial, waited, cleanup = asyncio.run(run())
+
+    assert "Process running" in initial
+    assert "booting" in initial + waited
+    assert "ready" in waited
+    assert "Wait target not observed" not in waited
+    assert "Session terminated." in cleanup
+
+
+def test_write_stdin_wait_for_reports_timeout_without_killing_session(tmp_path):
+    async def run() -> tuple[str, str, str]:
+        manager = ExecSessionManager()
+        exec_tool = ExecTool(working_dir=str(tmp_path), timeout=5, session_manager=manager)
+        stdin_tool = WriteStdinTool(manager=manager)
+        command = _python_command(
+            "import time; print('booting', flush=True); time.sleep(5)"
+        )
+
+        initial = await exec_tool.execute(command=command, yield_time_ms=100)
+        sid = _session_id(initial)
+        waited = await stdin_tool.execute(
+            session_id=sid,
+            wait_for="never-ready",
+            wait_timeout_ms=200,
+            yield_time_ms=0,
+        )
+        cleanup = await stdin_tool.execute(session_id=sid, terminate=True, yield_time_ms=0)
+        return initial, waited, cleanup
+
+    initial, waited, cleanup = asyncio.run(run())
+
+    assert "Process running" in initial
+    assert "booting" in initial + waited
+    assert "Process running" in waited
+    assert "Wait target not observed: 'never-ready'" in waited
+    assert "Session terminated." in cleanup
+
+
+def test_exec_session_mode_reuses_exec_safety_guard(tmp_path):
+    manager = ExecSessionManager()
+    tool = ExecTool(
+        working_dir=str(tmp_path),
+        deny_patterns=[r"echo\s+blocked"],
+        session_manager=manager,
+    )
+
+    result = asyncio.run(tool.execute(command="echo blocked", yield_time_ms=0))
+
+    assert "blocked by deny pattern" in result
+
+
+def test_write_stdin_reports_missing_session(tmp_path):
+    manager = ExecSessionManager()
+    tool = WriteStdinTool(manager=manager)
+
+    result = asyncio.run(tool.execute(session_id="missing", chars=""))
+
+    assert "exec session not found" in result
+
+
+def test_list_exec_sessions_reports_running_commands(tmp_path):
+    async def run() -> tuple[str, str, str]:
+        manager = ExecSessionManager()
+        exec_tool = ExecTool(working_dir=str(tmp_path), timeout=5, session_manager=manager)
+        list_tool = ListExecSessionsTool(manager=manager)
+        stdin_tool = WriteStdinTool(manager=manager)
+        command = _python_command(
+            "import time; print('ready', flush=True); time.sleep(5)"
+        )
+
+        initial = await exec_tool.execute(command=command, yield_time_ms=500)
+        sid = _session_id(initial)
+        listing = await list_tool.execute()
+        cleanup = await stdin_tool.execute(session_id=sid, terminate=True, yield_time_ms=0)
+        return sid, listing, cleanup
+
+    sid, listing, cleanup = asyncio.run(run())
+
+    assert sid in listing
+    assert "running" in listing
+    assert "elapsed=" in listing
+    assert "remaining=" in listing
+    assert str(tmp_path) in listing
+    assert "Session terminated." in cleanup
+
+
+def test_list_exec_sessions_reports_empty_state():
+    result = asyncio.run(ListExecSessionsTool(manager=ExecSessionManager()).execute())
+
+    assert result == "No active exec sessions."

tests/tools/test_file_edit_coding_enhancements.py

@@ -0,0 +1,216 @@
+from __future__ import annotations
+
+import asyncio
+
+from nanobot.agent.tools.filesystem import EditFileTool, ReadFileTool
+
+
+def test_read_file_force_bypasses_dedup(tmp_path):
+    target = tmp_path / "data.txt"
+    target.write_text("alpha\n")
+    tool = ReadFileTool(workspace=tmp_path)
+
+    first = asyncio.run(tool.execute(path=str(target)))
+    second = asyncio.run(tool.execute(path=str(target)))
+    forced = asyncio.run(tool.execute(path=str(target), force=True))
+
+    assert "alpha" in first
+    assert "unchanged" in second.lower()
+    assert "alpha" in forced
+    assert "unchanged" not in forced.lower()
+
+
+def test_edit_file_can_select_occurrence(tmp_path):
+    target = tmp_path / "duplicate.txt"
+    target.write_text("one\nsame\ntwo\nsame\n")
+    tool = EditFileTool(workspace=tmp_path)
+
+    result = asyncio.run(tool.execute(
+        path=str(target),
+        old_text="same",
+        new_text="changed",
+        occurrence=2,
+    ))
+
+    assert "Successfully edited" in result
+    assert target.read_text() == "one\nsame\ntwo\nchanged\n"
+
+
+def test_edit_file_expected_replacements_guards_replace_all(tmp_path):
+    target = tmp_path / "duplicate.txt"
+    target.write_text("same\nsame\n")
+    tool = EditFileTool(workspace=tmp_path)
+
+    result = asyncio.run(tool.execute(
+        path=str(target),
+        old_text="same",
+        new_text="changed",
+        replace_all=True,
+        expected_replacements=1,
+    ))
+
+    assert "expected 1 replacements but would make 2" in result
+    assert target.read_text() == "same\nsame\n"
+
+
+def test_edit_file_expected_replacements_allows_replace_all_when_count_matches(tmp_path):
+    target = tmp_path / "duplicate.txt"
+    target.write_text("same\nsame\n")
+    tool = EditFileTool(workspace=tmp_path)
+
+    result = asyncio.run(tool.execute(
+        path=str(target),
+        old_text="same",
+        new_text="changed",
+        replace_all=True,
+        expected_replacements=2,
+    ))
+
+    assert "Successfully edited" in result
+    assert target.read_text() == "changed\nchanged\n"
+
+
+def test_edit_file_can_select_nearest_line_hint(tmp_path):
+    target = tmp_path / "duplicate.txt"
+    target.write_text("one\nsame\ntwo\nsame\n")
+    tool = EditFileTool(workspace=tmp_path)
+
+    result = asyncio.run(tool.execute(
+        path=str(target),
+        old_text="same",
+        new_text="changed",
+        line_hint=4,
+    ))
+
+    assert "Successfully edited" in result
+    assert target.read_text() == "one\nsame\ntwo\nchanged\n"
+
+
+def test_edit_file_can_edit_ipynb_as_json(tmp_path):
+    target = tmp_path / "analysis.ipynb"
+    target.write_text('{"cells": []}')
+    tool = EditFileTool(workspace=tmp_path)
+
+    result = asyncio.run(tool.execute(
+        path=str(target),
+        old_text='"cells": []',
+        new_text='"cells": [{"cell_type": "markdown", "source": "hi"}]',
+    ))
+
+    assert "Successfully edited" in result
+    assert '"source": "hi"' in target.read_text()
+
+
+def test_edit_file_multiple_match_hint_mentions_occurrence(tmp_path):
+    target = tmp_path / "duplicate.txt"
+    target.write_text("same\nsame\n")
+    tool = EditFileTool(workspace=tmp_path)
+
+    result = asyncio.run(tool.execute(
+        path=str(target),
+        old_text="same",
+        new_text="changed",
+    ))
+
+    assert "old_text appears 2 times" in result
+    assert "occurrence" in result
+    assert target.read_text() == "same\nsame\n"
+
+
+def test_edit_file_rejects_ambiguous_line_hint(tmp_path):
+    target = tmp_path / "duplicate.txt"
+    target.write_text("same\nmiddle\nsame\n")
+    tool = EditFileTool(workspace=tmp_path)
+
+    result = asyncio.run(tool.execute(
+        path=str(target),
+        old_text="same",
+        new_text="changed",
+        line_hint=2,
+    ))
+
+    assert "line_hint 2 is ambiguous" in result
+    assert target.read_text() == "same\nmiddle\nsame\n"
+
+
+def test_edit_file_rejects_occurrence_with_replace_all(tmp_path):
+    target = tmp_path / "duplicate.txt"
+    target.write_text("same\nsame\n")
+    tool = EditFileTool(workspace=tmp_path)
+
+    result = asyncio.run(tool.execute(
+        path=str(target),
+        old_text="same",
+        new_text="changed",
+        occurrence=1,
+        replace_all=True,
+    ))
+
+    assert "occurrence cannot be used with replace_all" in result
+    assert target.read_text() == "same\nsame\n"
+
+
+def test_edit_file_rejects_line_hint_with_replace_all(tmp_path):
+    target = tmp_path / "duplicate.txt"
+    target.write_text("same\nsame\n")
+    tool = EditFileTool(workspace=tmp_path)
+
+    result = asyncio.run(tool.execute(
+        path=str(target),
+        old_text="same",
+        new_text="changed",
+        line_hint=1,
+        replace_all=True,
+    ))
+
+    assert "line_hint cannot be used with replace_all" in result
+    assert target.read_text() == "same\nsame\n"
+
+
+def test_edit_file_rejects_line_hint_with_occurrence(tmp_path):
+    target = tmp_path / "duplicate.txt"
+    target.write_text("same\nsame\n")
+    tool = EditFileTool(workspace=tmp_path)
+
+    result = asyncio.run(tool.execute(
+        path=str(target),
+        old_text="same",
+        new_text="changed",
+        occurrence=1,
+        line_hint=1,
+    ))
+
+    assert "line_hint cannot be used with occurrence" in result
+    assert target.read_text() == "same\nsame\n"
+
+
+def test_edit_file_rejects_zero_occurrence(tmp_path):
+    target = tmp_path / "duplicate.txt"
+    target.write_text("same\n")
+    tool = EditFileTool(workspace=tmp_path)
+
+    result = asyncio.run(tool.execute(
+        path=str(target),
+        old_text="same",
+        new_text="changed",
+        occurrence=0,
+    ))
+
+    assert "occurrence must be >= 1" in result
+    assert target.read_text() == "same\n"
+
+
+def test_edit_file_rejects_zero_line_hint(tmp_path):
+    target = tmp_path / "duplicate.txt"
+    target.write_text("same\n")
+    tool = EditFileTool(workspace=tmp_path)
+
+    result = asyncio.run(tool.execute(
+        path=str(target),
+        old_text="same",
+        new_text="changed",
+        line_hint=0,
+    ))
+
+    assert "line_hint must be >= 1" in result
+    assert target.read_text() == "same\n"

tests/tools/test_notebook_tool.py

@@ -1,147 +0,0 @@
-"""Tests for NotebookEditTool — Jupyter .ipynb editing."""
-
-import json
-
-import pytest
-
-from nanobot.agent.tools.notebook import NotebookEditTool
-
-
-def _make_notebook(cells: list[dict] | None = None, nbformat: int = 4, nbformat_minor: int = 5) -> dict:
-    """Build a minimal valid .ipynb structure."""
-    return {
-        "nbformat": nbformat,
-        "nbformat_minor": nbformat_minor,
-        "metadata": {"kernelspec": {"display_name": "Python 3", "language": "python", "name": "python3"}},
-        "cells": cells or [],
-    }
-
-
-def _code_cell(source: str, cell_id: str | None = None) -> dict:
-    cell = {"cell_type": "code", "source": source, "metadata": {}, "outputs": [], "execution_count": None}
-    if cell_id:
-        cell["id"] = cell_id
-    return cell
-
-
-def _md_cell(source: str, cell_id: str | None = None) -> dict:
-    cell = {"cell_type": "markdown", "source": source, "metadata": {}}
-    if cell_id:
-        cell["id"] = cell_id
-    return cell
-
-
-def _write_nb(tmp_path, name: str, nb: dict) -> str:
-    p = tmp_path / name
-    p.write_text(json.dumps(nb), encoding="utf-8")
-    return str(p)
-
-
-class TestNotebookEdit:
-
-    @pytest.fixture()
-    def tool(self, tmp_path):
-        return NotebookEditTool(workspace=tmp_path)
-
-    @pytest.mark.asyncio
-    async def test_replace_cell_content(self, tool, tmp_path):
-        nb = _make_notebook([_code_cell("print('hello')"), _code_cell("x = 1")])
-        path = _write_nb(tmp_path, "test.ipynb", nb)
-        result = await tool.execute(path=path, cell_index=0, new_source="print('world')")
-        assert "Successfully" in result
-        saved = json.loads((tmp_path / "test.ipynb").read_text())
-        assert saved["cells"][0]["source"] == "print('world')"
-        assert saved["cells"][1]["source"] == "x = 1"
-
-    @pytest.mark.asyncio
-    async def test_insert_cell_after_target(self, tool, tmp_path):
-        nb = _make_notebook([_code_cell("cell 0"), _code_cell("cell 1")])
-        path = _write_nb(tmp_path, "test.ipynb", nb)
-        result = await tool.execute(path=path, cell_index=0, new_source="inserted", edit_mode="insert")
-        assert "Successfully" in result
-        saved = json.loads((tmp_path / "test.ipynb").read_text())
-        assert len(saved["cells"]) == 3
-        assert saved["cells"][0]["source"] == "cell 0"
-        assert saved["cells"][1]["source"] == "inserted"
-        assert saved["cells"][2]["source"] == "cell 1"
-
-    @pytest.mark.asyncio
-    async def test_delete_cell(self, tool, tmp_path):
-        nb = _make_notebook([_code_cell("A"), _code_cell("B"), _code_cell("C")])
-        path = _write_nb(tmp_path, "test.ipynb", nb)
-        result = await tool.execute(path=path, cell_index=1, edit_mode="delete")
-        assert "Successfully" in result
-        saved = json.loads((tmp_path / "test.ipynb").read_text())
-        assert len(saved["cells"]) == 2
-        assert saved["cells"][0]["source"] == "A"
-        assert saved["cells"][1]["source"] == "C"
-
-    @pytest.mark.asyncio
-    async def test_create_new_notebook_from_scratch(self, tool, tmp_path):
-        path = str(tmp_path / "new.ipynb")
-        result = await tool.execute(path=path, cell_index=0, new_source="# Hello", edit_mode="insert", cell_type="markdown")
-        assert "Successfully" in result or "created" in result.lower()
-        saved = json.loads((tmp_path / "new.ipynb").read_text())
-        assert saved["nbformat"] == 4
-        assert len(saved["cells"]) == 1
-        assert saved["cells"][0]["cell_type"] == "markdown"
-        assert saved["cells"][0]["source"] == "# Hello"
-
-    @pytest.mark.asyncio
-    async def test_invalid_cell_index_error(self, tool, tmp_path):
-        nb = _make_notebook([_code_cell("only cell")])
-        path = _write_nb(tmp_path, "test.ipynb", nb)
-        result = await tool.execute(path=path, cell_index=5, new_source="x")
-        assert "Error" in result
-
-    @pytest.mark.asyncio
-    async def test_non_ipynb_rejected(self, tool, tmp_path):
-        f = tmp_path / "script.py"
-        f.write_text("pass")
-        result = await tool.execute(path=str(f), cell_index=0, new_source="x")
-        assert "Error" in result
-        assert ".ipynb" in result
-
-    @pytest.mark.asyncio
-    async def test_preserves_metadata_and_outputs(self, tool, tmp_path):
-        cell = _code_cell("old")
-        cell["outputs"] = [{"output_type": "stream", "text": "hello\n"}]
-        cell["execution_count"] = 42
-        nb = _make_notebook([cell])
-        path = _write_nb(tmp_path, "test.ipynb", nb)
-        await tool.execute(path=path, cell_index=0, new_source="new")
-        saved = json.loads((tmp_path / "test.ipynb").read_text())
-        assert saved["metadata"]["kernelspec"]["language"] == "python"
-
-    @pytest.mark.asyncio
-    async def test_nbformat_45_generates_cell_id(self, tool, tmp_path):
-        nb = _make_notebook([], nbformat_minor=5)
-        path = _write_nb(tmp_path, "test.ipynb", nb)
-        await tool.execute(path=path, cell_index=0, new_source="x = 1", edit_mode="insert")
-        saved = json.loads((tmp_path / "test.ipynb").read_text())
-        assert "id" in saved["cells"][0]
-        assert len(saved["cells"][0]["id"]) > 0
-
-    @pytest.mark.asyncio
-    async def test_insert_with_cell_type_markdown(self, tool, tmp_path):
-        nb = _make_notebook([_code_cell("code")])
-        path = _write_nb(tmp_path, "test.ipynb", nb)
-        await tool.execute(path=path, cell_index=0, new_source="# Title", edit_mode="insert", cell_type="markdown")
-        saved = json.loads((tmp_path / "test.ipynb").read_text())
-        assert saved["cells"][1]["cell_type"] == "markdown"
-
-    @pytest.mark.asyncio
-    async def test_invalid_edit_mode_rejected(self, tool, tmp_path):
-        nb = _make_notebook([_code_cell("code")])
-        path = _write_nb(tmp_path, "test.ipynb", nb)
-        result = await tool.execute(path=path, cell_index=0, new_source="x", edit_mode="replcae")
-        assert "Error" in result
-        assert "edit_mode" in result
-
-    @pytest.mark.asyncio
-    async def test_invalid_cell_type_rejected(self, tool, tmp_path):
-        nb = _make_notebook([_code_cell("code")])
-        path = _write_nb(tmp_path, "test.ipynb", nb)
-        result = await tool.execute(path=path, cell_index=0, new_source="x", cell_type="raw")
-        assert "Error" in result
-        assert "cell_type" in result

tests/tools/test_search_tools.py

@@ -12,7 +12,7 @@ import pytest
 
 from nanobot.agent.loop import AgentLoop
 from nanobot.agent.subagent import SubagentManager, SubagentStatus
-from nanobot.agent.tools.search import GrepTool
+from nanobot.agent.tools.search import FindFilesTool, GrepTool
 from nanobot.agent.tools.web import WebSearchTool
 from nanobot.bus.queue import MessageBus
 from nanobot.config.schema import WebSearchConfig
@@ -33,6 +33,68 @@ async def test_web_search_tool_refreshes_dynamic_config_loader(monkeypatch) -> N
     assert await tool.execute("nanobot") == "duckduckgo:nanobot:3"
 
 
+@pytest.mark.asyncio
+async def test_find_files_filters_by_query_glob_and_type(tmp_path: Path) -> None:
+    (tmp_path / "src").mkdir()
+    (tmp_path / "src" / "settings_view.tsx").write_text("export {}\n", encoding="utf-8")
+    (tmp_path / "src" / "settings_api.py").write_text("pass\n", encoding="utf-8")
+    (tmp_path / "README.md").write_text("settings\n", encoding="utf-8")
+
+    tool = FindFilesTool(workspace=tmp_path, allowed_dir=tmp_path)
+    result = await tool.execute(
+        path=".",
+        query="settings",
+        glob="src/**",
+        type="ts",
+    )
+
+    assert result.splitlines() == ["src/settings_view.tsx"]
+
+
+@pytest.mark.asyncio
+async def test_find_files_can_include_directories(tmp_path: Path) -> None:
+    (tmp_path / "src" / "settings").mkdir(parents=True)
+    (tmp_path / "src" / "settings" / "index.ts").write_text("export {}\n", encoding="utf-8")
+
+    tool = FindFilesTool(workspace=tmp_path, allowed_dir=tmp_path)
+    result = await tool.execute(path="src", query="settings", include_dirs=True)
+
+    assert "src/settings/" in result.splitlines()
+    assert "src/settings/index.ts" in result.splitlines()
+
+
+@pytest.mark.asyncio
+async def test_find_files_supports_modified_sort_and_pagination(tmp_path: Path) -> None:
+    (tmp_path / "src").mkdir()
+    for idx, name in enumerate(("a.py", "b.py", "c.py"), start=1):
+        file_path = tmp_path / "src" / name
+        file_path.write_text("pass\n", encoding="utf-8")
+        os.utime(file_path, (idx, idx))
+
+    tool = FindFilesTool(workspace=tmp_path, allowed_dir=tmp_path)
+    result = await tool.execute(
+        path="src",
+        type="py",
+        sort="modified",
+        head_limit=1,
+        offset=1,
+    )
+
+    assert result.splitlines()[0] == "src/b.py"
+    assert "pagination: limit=1, offset=1" in result
+
+
+@pytest.mark.asyncio
+async def test_find_files_rejects_paths_outside_workspace(tmp_path: Path) -> None:
+    outside = tmp_path.parent / "outside-find-files.txt"
+    outside.write_text("secret\n", encoding="utf-8")
+
+    tool = FindFilesTool(workspace=tmp_path, allowed_dir=tmp_path)
+    result = await tool.execute(path=str(outside))
+
+    assert result.startswith("Error:")
+
+
 @pytest.mark.asyncio
 async def test_grep_respects_glob_filter_and_context(tmp_path: Path) -> None:
     (tmp_path / "src").mkdir()
@@ -249,6 +311,7 @@ def test_agent_loop_registers_grep(tmp_path: Path) -> None:
 
     loop = AgentLoop(bus=bus, provider=provider, workspace=tmp_path, model="test-model")
 
+    assert "find_files" in loop.tools.tool_names
     assert "grep" in loop.tools.tool_names
 
 
@@ -280,6 +343,7 @@ async def test_subagent_registers_grep(tmp_path: Path) -> None:
     status = SubagentStatus(task_id="sub-1", label="label", task_description="search task", started_at=time.monotonic())
     await mgr._run_subagent("sub-1", "search task", "label", {"channel": "cli", "chat_id": "direct"}, status)
 
+    assert "find_files" in captured["tool_names"]
     assert "grep" in captured["tool_names"]
 
 

tests/tools/test_tool_descriptions.py

@@ -0,0 +1,46 @@
+from nanobot.agent.tools.apply_patch import ApplyPatchTool
+from nanobot.agent.tools.exec_session import ListExecSessionsTool, WriteStdinTool
+from nanobot.agent.tools.filesystem import EditFileTool, ReadFileTool, WriteFileTool
+from nanobot.agent.tools.search import FindFilesTool, GrepTool
+from nanobot.agent.tools.shell import ExecTool
+
+
+def test_coding_tool_descriptions_steer_editing_priority() -> None:
+    apply_patch = ApplyPatchTool().description.lower()
+    edit_file = EditFileTool().description.lower()
+    write_file = WriteFileTool().description.lower()
+
+    assert "default tool for code edits" in apply_patch
+    assert "multi-file" in apply_patch
+    assert "dry_run=true" in apply_patch
+    assert "edit_file only for small exact replacements" in apply_patch
+
+    assert "small, exact replacement" in edit_file
+    assert "copied from read_file" in edit_file
+    assert "prefer apply_patch" in edit_file
+
+    assert "replace an entire file" in write_file
+    assert "prefer apply_patch" in write_file
+
+
+def test_coding_tool_descriptions_steer_discovery_and_shell_usage() -> None:
+    read_file = ReadFileTool().description.lower()
+    find_files = FindFilesTool().description.lower()
+    grep = GrepTool().description.lower()
+    exec_tool = ExecTool().description.lower()
+    write_stdin = WriteStdinTool().description.lower()
+    list_sessions = ListExecSessionsTool().description.lower()
+
+    assert "find_files/list_dir first" in read_file
+    assert "before editing" in read_file
+    assert "prefer it over shell find/ls" in find_files
+    assert "prefer this over shell grep" in grep
+
+    assert "tests, builds" in exec_tool
+    assert "prefer read_file/find_files/grep" in exec_tool
+    assert "apply_patch/write_file/edit_file" in exec_tool
+    assert "yield_time_ms" in exec_tool
+
+    assert "do not use this to start new commands" in write_stdin
+    assert "wait_for" in write_stdin
+    assert "recover a session_id" in list_sessions

tests/tools/test_tool_loader.py

@@ -89,9 +89,11 @@ def test_discover_finds_concrete_tools():
     loader = ToolLoader()
     discovered = loader.discover()
     class_names = {cls.__name__ for cls in discovered}
+    assert "ApplyPatchTool" in class_names
     assert "ExecTool" in class_names
     assert "MessageTool" in class_names
     assert "SpawnTool" in class_names
+    assert "WriteStdinTool" in class_names
 
 
 def test_discover_excludes_abstract_and_mcp():
@@ -406,7 +408,8 @@ def test_loader_registers_same_tools_as_old_hardcoded():
 
     expected = {
         "read_file", "write_file", "edit_file", "list_dir",
-        "grep", "notebook_edit", "exec", "web_search", "web_fetch",
+        "find_files", "grep", "exec", "write_stdin", "list_exec_sessions",
+        "web_search", "web_fetch",
         "message", "spawn", "cron",
     }
     actual = set(registered)

tests/tools/test_tool_validation.py

@@ -3,6 +3,8 @@ import subprocess
 import sys
 from typing import Any
 
+import pytest
+
 from nanobot.agent.tools import (
     ArraySchema,
     IntegerSchema,
@@ -15,6 +17,7 @@ from nanobot.agent.tools import (
 from nanobot.agent.tools.base import Tool
 from nanobot.agent.tools.registry import ToolRegistry
 from nanobot.agent.tools.shell import ExecTool
+from nanobot.security.network import configure_ssrf_whitelist
 
 
 class SampleTool(Tool):
@@ -218,6 +221,39 @@ def test_exec_extract_absolute_paths_ignores_relative_posix_segments() -> None:
     assert "/bin/python" not in paths
 
 
+def test_exec_extract_absolute_paths_ignores_urls() -> None:
+    cmd = 'curl -s -o /dev/null -w "%{http_code}" https://www.google.com'
+    paths = ExecTool._extract_absolute_paths(cmd)
+    assert paths == ["/dev/null"]
+
+
+@pytest.mark.parametrize(
+    "command",
+    [
+        'curl -s -o /dev/null -w "%{http_code}" https://www.google.com',
+        'wget -q -O - http://example.com 2>&1 | head -c 100',
+        'python3 -c "import urllib.request; print(urllib.request.urlopen(\'http://example.com\').read()[:100])"',
+    ],
+)
+def test_exec_guard_allows_public_urls(tmp_path, command: str) -> None:
+    tool = ExecTool(restrict_to_workspace=True)
+    error = tool._guard_command(command, str(tmp_path))
+    assert error is None
+
+
+def test_exec_guard_allows_whitelisted_internal_urls(tmp_path) -> None:
+    configure_ssrf_whitelist(["10.10.10.0/24"])
+    try:
+        tool = ExecTool(restrict_to_workspace=True)
+        error = tool._guard_command(
+            'curl -s -H "Authorization: Bearer ..." http://10.10.10.3:8123/api/',
+            str(tmp_path),
+        )
+        assert error is None
+    finally:
+        configure_ssrf_whitelist([])
+
+
 def test_exec_extract_absolute_paths_captures_posix_absolute_paths() -> None:
     cmd = "cat /tmp/data.txt > /tmp/out.txt"
     paths = ExecTool._extract_absolute_paths(cmd)