feat(tools): tighten patch and session workflows

tests/tools/test_apply_patch_tool.py

@@ -40,9 +40,58 @@ def test_apply_patch_updates_multiple_hunks(tmp_path):
     ))
 
     assert "update multi.txt" in result
+    assert "(+2/-2)" in result
     assert target.read_text() == "line1\nchanged2\nline3\nchanged4\n"
 
 
+def test_apply_patch_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(
+        patch="""*** Begin Patch
+*** Update File: dry.txt
+@@
+-before
++after
+*** Add File: added.txt
++new
+*** End Patch
+""",
+        dry_run=True,
+    ))
+
+    assert "Patch dry-run succeeded" in result
+    assert "- update dry.txt (+1/-1)" in result
+    assert "- add added.txt (+1/-0)" in result
+    assert target.read_text() == "before\n"
+    assert not (tmp_path / "added.txt").exists()
+
+
+def test_apply_patch_applies_repeated_update_sections_sequentially(tmp_path):
+    target = tmp_path / "repeat.txt"
+    target.write_text("one\ntwo\nthree\n")
+    tool = ApplyPatchTool(workspace=tmp_path)
+
+    result = asyncio.run(tool.execute(
+        patch="""*** Begin Patch
+*** Update File: repeat.txt
+@@
+-one
++ONE
+*** Update File: repeat.txt
+@@
+-three
++THREE
+*** End Patch
+"""
+    ))
+
+    assert result.count("update repeat.txt") == 2
+    assert target.read_text() == "ONE\ntwo\nTHREE\n"
+
+
 def test_apply_patch_ignores_standard_no_newline_marker(tmp_path):
     target = tmp_path / "plain.txt"
     target.write_text("before")

tests/tools/test_exec_session_tools.py

@@ -245,6 +245,65 @@ def test_write_stdin_preserves_completed_session_output_until_polled(tmp_path):
     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 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 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(

tests/utils/test_file_edit_events.py

@@ -125,6 +125,28 @@ def test_apply_patch_prepares_trackers_for_each_touched_file(tmp_path: Path) ->
     assert (by_path["src/delete_me.py"]["added"], by_path["src/delete_me.py"]["deleted"]) == (0, 1)
 
 
+def test_apply_patch_dry_run_does_not_prepare_file_edit_trackers(tmp_path: Path) -> None:
+    (tmp_path / "file.txt").write_text("old\n", encoding="utf-8")
+
+    trackers = prepare_file_edit_trackers(
+        call_id="call-patch",
+        tool_name="apply_patch",
+        tool=None,
+        workspace=tmp_path,
+        params={
+            "dry_run": True,
+            "patch": """*** Begin Patch
+*** Update File: file.txt
+@@
+-old
++new
+*** End Patch""",
+        },
+    )
+
+    assert trackers == []
+
+
 def test_oversized_write_file_end_uses_known_content_for_exact_count(tmp_path: Path) -> None:
     target = tmp_path / "large.txt"
     params = {"path": "large.txt", "content": "x" * (2 * 1024 * 1024 + 1)}