diff --git a/python/packages/core/agent_framework/_harness/_file_access.py b/python/packages/core/agent_framework/_harness/_file_access.py index 7dc95779eb..9b9901fc55 100644 --- a/python/packages/core/agent_framework/_harness/_file_access.py +++ b/python/packages/core/agent_framework/_harness/_file_access.py @@ -27,6 +27,7 @@ import logging import os import re +import threading from abc import ABC, abstractmethod from collections.abc import Awaitable, Mapping, MutableMapping from pathlib import Path @@ -1097,6 +1098,8 @@ class FileSystemAgentFileStore(AgentFileStore): hostile process that shares the root directory. """ + _DELETE_LOCK: ClassVar[threading.Lock] = threading.Lock() + def __init__(self, root_directory: str | os.PathLike[str]) -> None: """Initialize the file-system store. @@ -1284,11 +1287,15 @@ async def delete(self, path: str) -> bool: full_path = self._resolve_safe_path(path) return await asyncio.to_thread(self._delete_file_sync, full_path) - @staticmethod - def _delete_file_sync(full_path: Path) -> bool: - if not full_path.is_file(): - return False - full_path.unlink() + @classmethod + def _delete_file_sync(cls, full_path: Path) -> bool: + with cls._DELETE_LOCK: + if not full_path.is_file(): + return False + try: + full_path.unlink() + except FileNotFoundError: + return False return True async def list_children(self, directory: str = "") -> list[FileStoreEntry]: diff --git a/python/packages/core/tests/core/test_harness_file_access.py b/python/packages/core/tests/core/test_harness_file_access.py index da7b4255f4..030b80f151 100644 --- a/python/packages/core/tests/core/test_harness_file_access.py +++ b/python/packages/core/tests/core/test_harness_file_access.py @@ -3,13 +3,16 @@ from __future__ import annotations import asyncio +import itertools import json import os import re import stat +import threading import time from pathlib import Path from types import SimpleNamespace +from typing import Any import pytest @@ -287,6 +290,99 @@ async def test_filesystem_store_round_trips_files(tmp_path: Path) -> None: assert await store.delete("nested/a.txt") is False +async def _run_deterministic_concurrent_deletes( + store: FileSystemAgentFileStore, + other_store: FileSystemAgentFileStore, + first_path: str, + second_path: str, + monkeypatch: pytest.MonkeyPatch, +) -> tuple[bool, bool]: + target_paths = {store._resolve_safe_path(first_path), other_store._resolve_safe_path(second_path)} + original_to_thread = asyncio.to_thread + original_unlink = Path.unlink + worker_barrier = threading.Barrier(2) + unlink_barrier = threading.Barrier(2) + unlink_guard = threading.Lock() + overlapping_delete_completed = False + + async def synchronized_to_thread(function: Any, /, *args: Any, **kwargs: Any) -> Any: + def synchronized_call() -> Any: + worker_barrier.wait(timeout=5) + return function(*args, **kwargs) + + return await original_to_thread(synchronized_call) + + def macos_style_unlink(path: Path, missing_ok: bool = False) -> None: + nonlocal overlapping_delete_completed + if path not in target_paths: + original_unlink(path, missing_ok=missing_ok) + return + + try: + unlink_barrier.wait(timeout=0.5) + except threading.BrokenBarrierError: + original_unlink(path, missing_ok=missing_ok) + return + + # Model concurrent macOS unlinks, where both calls may report success even though only one removes the file. + with unlink_guard: + if not overlapping_delete_completed: + original_unlink(path, missing_ok=missing_ok) + overlapping_delete_completed = True + + monkeypatch.setattr(asyncio, "to_thread", synchronized_to_thread) + monkeypatch.setattr(Path, "unlink", macos_style_unlink) + return await asyncio.gather(store.delete(first_path), other_store.delete(second_path)) + + +async def test_filesystem_store_concurrent_delete(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Concurrent deletion should report one deletion and one missing file.""" + store = FileSystemAgentFileStore(tmp_path) + other_store = FileSystemAgentFileStore(tmp_path) + await store.write("shared.txt", "content") + + results = await _run_deterministic_concurrent_deletes(store, other_store, "shared.txt", "shared.txt", monkeypatch) + + assert sorted(results) == [False, True] + + +async def test_filesystem_store_concurrent_delete_case_aliases(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Concurrent deletion through case aliases should report one deletion.""" + store = FileSystemAgentFileStore(tmp_path) + other_store = FileSystemAgentFileStore(tmp_path) + original_name = "Shared.txt" + await store.write(original_name, "content") + + original_path = store._resolve_safe_path(original_name) + lower_path = tmp_path / original_name.lower() + if not await asyncio.to_thread(lower_path.exists) or not await asyncio.to_thread( + os.path.samefile, original_path, lower_path + ): + pytest.skip("filesystem does not treat ASCII case variants as aliases") + + # Choose a case alias with opposite hash parity so the regression + # deterministically exercises the former path-hash-keyed synchronization bug + # without depending on this interpreter's randomized hash values. + alias_name = next( + ( + candidate_name + for characters in itertools.product(*[ + (character.lower(), character.upper()) for character in original_name + ]) + if (candidate_name := "".join(characters)) != original_name + and hash(store._resolve_safe_path(candidate_name)) % 2 != hash(original_path) % 2 + ), + None, + ) + assert alias_name is not None + alias_path = other_store._resolve_safe_path(alias_name) + assert await asyncio.to_thread(os.path.samefile, original_path, alias_path) + + results = await _run_deterministic_concurrent_deletes(store, other_store, original_name, alias_name, monkeypatch) + + assert sorted(results) == [False, True] + + async def test_filesystem_store_rejects_traversal_and_rooted_paths(tmp_path: Path) -> None: """The filesystem store should refuse paths that escape the configured root.""" store = FileSystemAgentFileStore(tmp_path)