Description
Concurrent calls to FileCheckpointStorage.save() for the same checkpoint ID use the same temporary path:
tmp_path = file_path.with_suffix(".json.tmp")Each call writes to and subsequently renames that shared file. Once one call moves it to the final destination, competing calls fail with FileNotFoundError.
Expected behavior
Concurrent saves should not fail because of an internal temporary-file collision. Each operation should use its own temporary file, with the final successful replacement determining the stored checkpoint.
Actual behavior
Concurrent saves frequently fail when os.replace() cannot find the shared temporary file.
Code Sample
import asyncio
import tempfile
from pathlib import Path
from agent_framework import FileCheckpointStorage, WorkflowCheckpoint
async def main() -> None:
with tempfile.TemporaryDirectory() as directory:
storage = FileCheckpointStorage(Path(directory))
checkpoint = WorkflowCheckpoint(
workflow_name="test-workflow",
graph_signature_hash="test-hash",
checkpoint_id="shared-id",
)
results = await asyncio.gather(
*(storage.save(checkpoint) for _ in range(100)),
return_exceptions=True,
)
errors = [
result
for result in results
if isinstance(result, BaseException)
]
print(f"Failed saves: {len(errors)}")
print(f"First error: {errors[0]!r}")
asyncio.run(main())
In repeated runs, between 49 and 71 of the 100 saves failed.
## Error Messages / Stack Traces
FileNotFoundError: [Errno 2] No such file or directory:
'<storage>/shared-id.json.tmp' -> '<storage>/shared-id.json'
The exception originates from:
os.replace(tmp_path, file_path)Error Messages / Stack Traces
Package Versions
agent-framework-core: 1.13.0
Python Version
Python 3.12.10
Additional Context
FileCheckpointStorage.save() should create a unique temporary file for each operation in the destination directory and then atomically replace the final path.
A regression test should run multiple saves concurrently for the same checkpoint ID and verify that every save completes without an internal filesystem exception.
Description
Concurrent calls to
FileCheckpointStorage.save()for the same checkpoint ID use the same temporary path:Each call writes to and subsequently renames that shared file. Once one call moves it to the final destination, competing calls fail with
FileNotFoundError.Expected behavior
Concurrent saves should not fail because of an internal temporary-file collision. Each operation should use its own temporary file, with the final successful replacement determining the stored checkpoint.
Actual behavior
Concurrent saves frequently fail when
os.replace()cannot find the shared temporary file.Code Sample
import asyncio import tempfile from pathlib import Path from agent_framework import FileCheckpointStorage, WorkflowCheckpoint async def main() -> None: with tempfile.TemporaryDirectory() as directory: storage = FileCheckpointStorage(Path(directory)) checkpoint = WorkflowCheckpoint( workflow_name="test-workflow", graph_signature_hash="test-hash", checkpoint_id="shared-id", ) results = await asyncio.gather( *(storage.save(checkpoint) for _ in range(100)), return_exceptions=True, ) errors = [ result for result in results if isinstance(result, BaseException) ] print(f"Failed saves: {len(errors)}") print(f"First error: {errors[0]!r}") asyncio.run(main()) In repeated runs, between 49 and 71 of the 100 saves failed. ## Error Messages / Stack Traces FileNotFoundError: [Errno 2] No such file or directory: '<storage>/shared-id.json.tmp' -> '<storage>/shared-id.json' The exception originates from: os.replace(tmp_path, file_path)Error Messages / Stack Traces
Package Versions
agent-framework-core: 1.13.0
Python Version
Python 3.12.10
Additional Context
FileCheckpointStorage.save()should create a unique temporary file for each operation in the destination directory and then atomically replace the final path.A regression test should run multiple saves concurrently for the same checkpoint ID and verify that every save completes without an internal filesystem exception.