From 78ddc7ddc57a5475750b4bfdbc568ab6bbac0001 Mon Sep 17 00:00:00 2001 From: Yufeng He <40085740+he-yufeng@users.noreply.github.com> Date: Sat, 17 Oct 2026 05:15:56 +0800 Subject: [PATCH 1/4] Python: fix(core): reject checkpoint IDs that collide by letter case on save On case-insensitive filesystems (default Windows/macOS), Round-A.json and round-a.json are the same file. FileCheckpointStorage.save accepted both spellings, so the second save silently replaced the first checkpoint and reads of the first ID then returned the other workflow's state, with the first workflow's listing left empty. save now fails closed: when the destination exists under a differently-cased spelling, the save raises WorkflowCheckpointException naming the collision instead of replacing a different checkpoint's data. Case-sensitive filesystems skip the scan entirely (the variant path does not exist there), and re-saving the exact same ID stays a normal update everywhere. Tests: a conformance regression pins the rejection, the first checkpoint's survival, and the same-ID update path; red without the guard on a case-insensitive volume. --- .../agent_framework/_workflows/_checkpoint.py | 32 +++++++++++++ .../test_checkpoint_storage_conformance.py | 47 +++++++++++++++++++ 2 files changed, 79 insertions(+) diff --git a/python/packages/core/agent_framework/_workflows/_checkpoint.py b/python/packages/core/agent_framework/_workflows/_checkpoint.py index cd78725d6a1..209d3908572 100644 --- a/python/packages/core/agent_framework/_workflows/_checkpoint.py +++ b/python/packages/core/agent_framework/_workflows/_checkpoint.py @@ -571,6 +571,37 @@ def _validate_file_path(self, checkpoint_id: CheckpointID) -> Path: raise WorkflowCheckpointException(f"Invalid checkpoint ID: {checkpoint_id}") return file_path + @staticmethod + def _assert_no_case_variant_collision(file_path: Path, checkpoint_id: CheckpointID) -> None: + """Reject an ID that the filesystem cannot store as an independent file. + + On case-insensitive filesystems (the default on Windows and macOS), + ``Round-A.json`` and ``round-a.json`` name the same file, so a save under + the variant spelling would silently replace the other checkpoint and its + reads would then return the wrong workflow's state (#9444). On + case-sensitive filesystems the variant path simply does not exist, so + the directory scan below is skipped entirely there. + """ + if not file_path.exists(): + return + expected_name = file_path.name + try: + for sibling in file_path.parent.iterdir(): + if sibling.name.casefold() != expected_name.casefold(): + continue + if sibling.name != expected_name: + raise WorkflowCheckpointException( + f"Checkpoint ID {checkpoint_id!r} collides with the existing checkpoint " + f"{sibling.name.removesuffix('.json')!r}: on this filesystem both names " + "resolve to one file, and saving would replace that checkpoint's data. " + "Choose a checkpoint ID that differs by more than letter case." + ) + return + except OSError: + # A transient listing/read failure must not turn a routine save into a + # hard failure; the write below still fails loudly if the path is bad. + return + @classmethod def _delete_lock(cls, file_path: Path) -> threading.Lock: """Return the process-local deletion lock for a checkpoint file.""" @@ -592,6 +623,7 @@ async def save(self, checkpoint: WorkflowCheckpoint) -> CheckpointID: from ._checkpoint_encoding import decode_checkpoint_value, encode_checkpoint_value file_path = self._validate_file_path(checkpoint.checkpoint_id) + self._assert_no_case_variant_collision(file_path, checkpoint.checkpoint_id) checkpoint_dict = checkpoint.to_dict() # Fail at save time if encoding or restore validation fails (#8181). try: diff --git a/python/packages/core/tests/workflow/test_checkpoint_storage_conformance.py b/python/packages/core/tests/workflow/test_checkpoint_storage_conformance.py index e1ce83d81f8..30236cb153d 100644 --- a/python/packages/core/tests/workflow/test_checkpoint_storage_conformance.py +++ b/python/packages/core/tests/workflow/test_checkpoint_storage_conformance.py @@ -152,3 +152,50 @@ async def test_load_missing_id_raises(conformance_storage: CheckpointStorage) -> async def test_save_returns_id(conformance_storage: CheckpointStorage) -> None: checkpoint = _conformance_checkpoint() assert await conformance_storage.save(checkpoint) == checkpoint.checkpoint_id + + +def _filesystem_is_case_insensitive(path: Path) -> bool: + probe = path / "CkAsE.probe" + probe.write_text("x") + try: + return (path / "ckase.probe").exists() + finally: + probe.unlink() + + +@pytest.mark.skipif( + not _filesystem_is_case_insensitive(Path(__file__).parent), + reason="case-variant collision only occurs on case-insensitive filesystems", +) +async def test_file_save_rejects_case_variant_checkpoint_id(tmp_path: Path) -> None: + """A case-only variant ID must not silently replace the other checkpoint (#9444).""" + storage = FileCheckpointStorage(tmp_path / "checkpoints") + first = WorkflowCheckpoint( + workflow_name="first-workflow", + graph_signature_hash="topology-a", + checkpoint_id="Round-A", + state={"marker": "first"}, + ) + second = WorkflowCheckpoint( + workflow_name="second-workflow", + graph_signature_hash="topology-b", + checkpoint_id="round-a", + state={"marker": "second"}, + ) + + first_id = await storage.save(first) + with pytest.raises(WorkflowCheckpointException, match="collides"): + await storage.save(second) + + # The first checkpoint survives untouched: its data and listing are intact. + restored = await storage.load(first_id) + assert restored.checkpoint_id == "Round-A" + assert restored.workflow_name == "first-workflow" + assert restored.state["marker"] == "first" + listed = await storage.list_checkpoint_ids(workflow_name="first-workflow") + assert listed == ["Round-A"] + + # Re-saving the exact same ID remains a normal update, not a collision. + first.state["marker"] = "first-updated" + assert await storage.save(first) == "Round-A" + assert (await storage.load("Round-A")).state["marker"] == "first-updated" From 183ae7b717f0f5d0a3e709a9b10ebd547f51aee5 Mon Sep 17 00:00:00 2001 From: Yufeng He <40085740+he-yufeng@users.noreply.github.com> Date: Sat, 17 Oct 2026 17:30:22 +0800 Subject: [PATCH 2/4] Python: fix(core): harden case-collision guard against path canonicalization Address Copilot review on the #9444 fix: derive the expected filename from the requested checkpoint ID instead of the resolved path (Windows resolve() canonicalizes to on-disk casing, which would accept the colliding save), allow same-ID updates when the exact spelling already exists on case-sensitive volumes, and fail closed when the directory cannot be listed. Move the case-sensitivity probe into the test body on tmp_path so parallel xdist workers cannot race a shared collection-time probe. --- .../agent_framework/_workflows/_checkpoint.py | 43 ++++++++++++------- .../test_checkpoint_storage_conformance.py | 9 ++-- 2 files changed, 33 insertions(+), 19 deletions(-) diff --git a/python/packages/core/agent_framework/_workflows/_checkpoint.py b/python/packages/core/agent_framework/_workflows/_checkpoint.py index 209d3908572..d3259b5b71e 100644 --- a/python/packages/core/agent_framework/_workflows/_checkpoint.py +++ b/python/packages/core/agent_framework/_workflows/_checkpoint.py @@ -584,23 +584,36 @@ def _assert_no_case_variant_collision(file_path: Path, checkpoint_id: Checkpoint """ if not file_path.exists(): return - expected_name = file_path.name + # Derive the expected spelling from the requested ID, not the resolved + # path: on Windows Path.resolve() canonicalizes an existing file to its + # on-disk casing, so file_path.name would already be the colliding + # name and the check below would accept the overwrite (#9444). + expected_name = f"{checkpoint_id}.json" try: - for sibling in file_path.parent.iterdir(): - if sibling.name.casefold() != expected_name.casefold(): - continue - if sibling.name != expected_name: - raise WorkflowCheckpointException( - f"Checkpoint ID {checkpoint_id!r} collides with the existing checkpoint " - f"{sibling.name.removesuffix('.json')!r}: on this filesystem both names " - "resolve to one file, and saving would replace that checkpoint's data. " - "Choose a checkpoint ID that differs by more than letter case." - ) - return - except OSError: - # A transient listing/read failure must not turn a routine save into a - # hard failure; the write below still fails loudly if the path is bad. + siblings = [entry.name for entry in file_path.parent.iterdir()] + except OSError as exc: + # Permissions can allow lookup/creation/replacement while denying + # enumeration, so skipping the scan here would let a case-variant + # save proceed unchecked. Fail closed instead. + raise WorkflowCheckpointException( + f"Cannot verify checkpoint ID {checkpoint_id!r} against existing checkpoints: " + "the storage directory could not be listed." + ) from exc + variant = next( + (name for name in siblings if name != expected_name and name.casefold() == expected_name.casefold()), + None, + ) + if variant is None or expected_name in siblings: + # No case-only variant exists, or the exact spelling is already + # stored (a same-ID update on a case-sensitive volume where both + # spellings legitimately coexist). return + raise WorkflowCheckpointException( + f"Checkpoint ID {checkpoint_id!r} collides with the existing checkpoint " + f"{variant.removesuffix('.json')!r}: on this filesystem both names " + "resolve to one file, and saving would replace that checkpoint's data. " + "Choose a checkpoint ID that differs by more than letter case." + ) @classmethod def _delete_lock(cls, file_path: Path) -> threading.Lock: diff --git a/python/packages/core/tests/workflow/test_checkpoint_storage_conformance.py b/python/packages/core/tests/workflow/test_checkpoint_storage_conformance.py index 30236cb153d..8c780b32624 100644 --- a/python/packages/core/tests/workflow/test_checkpoint_storage_conformance.py +++ b/python/packages/core/tests/workflow/test_checkpoint_storage_conformance.py @@ -163,12 +163,13 @@ def _filesystem_is_case_insensitive(path: Path) -> bool: probe.unlink() -@pytest.mark.skipif( - not _filesystem_is_case_insensitive(Path(__file__).parent), - reason="case-variant collision only occurs on case-insensitive filesystems", -) async def test_file_save_rejects_case_variant_checkpoint_id(tmp_path: Path) -> None: """A case-only variant ID must not silently replace the other checkpoint (#9444).""" + # Probe inside the test on tmp_path: a module-level probe races under + # pytest-xdist workers sharing the source directory, and tmp_path is the + # volume that actually holds the checkpoints. + if not _filesystem_is_case_insensitive(tmp_path): + pytest.skip("case-variant collision only occurs on case-insensitive filesystems") storage = FileCheckpointStorage(tmp_path / "checkpoints") first = WorkflowCheckpoint( workflow_name="first-workflow", From cddc81692a5e758171ff828a8d11a7174b06cc5c Mon Sep 17 00:00:00 2001 From: Yufeng He <40085740+he-yufeng@users.noreply.github.com> Date: Sun, 18 Oct 2026 05:50:44 +0800 Subject: [PATCH 3/4] Python: fix(core): fold Unicode normalization into the checkpoint collision guard APFS and HFS+ treat NFC and NFD spellings of a name as one file, and str.casefold() does not normalize, so a checkpoint ID differing only by normalization form still reached the silent overwrite from #9444. Compare siblings under NFC + casefold and reject those variants too. Thanks to Somay-kousis for the on-device report and the fix sketch. --- .../agent_framework/_workflows/_checkpoint.py | 32 ++++++---- .../test_checkpoint_storage_conformance.py | 58 +++++++++++++++++++ 2 files changed, 80 insertions(+), 10 deletions(-) diff --git a/python/packages/core/agent_framework/_workflows/_checkpoint.py b/python/packages/core/agent_framework/_workflows/_checkpoint.py index d3259b5b71e..6392e791b10 100644 --- a/python/packages/core/agent_framework/_workflows/_checkpoint.py +++ b/python/packages/core/agent_framework/_workflows/_checkpoint.py @@ -10,6 +10,7 @@ import os import threading import time +import unicodedata import uuid from collections.abc import Callable, Mapping from concurrent.futures import Future @@ -572,15 +573,17 @@ def _validate_file_path(self, checkpoint_id: CheckpointID) -> Path: return file_path @staticmethod - def _assert_no_case_variant_collision(file_path: Path, checkpoint_id: CheckpointID) -> None: + def _assert_no_variant_collision(file_path: Path, checkpoint_id: CheckpointID) -> None: """Reject an ID that the filesystem cannot store as an independent file. On case-insensitive filesystems (the default on Windows and macOS), ``Round-A.json`` and ``round-a.json`` name the same file, so a save under the variant spelling would silently replace the other checkpoint and its - reads would then return the wrong workflow's state (#9444). On - case-sensitive filesystems the variant path simply does not exist, so - the directory scan below is skipped entirely there. + reads would then return the wrong workflow's state (#9444). APFS and HFS+ + additionally fold Unicode normalization forms, so ``café-1.json`` spelled + NFC and NFD is also one file there (#9466 review). On filesystems without + those folds the variant path simply does not exist, so the directory scan + below is skipped entirely there. """ if not file_path.exists(): return @@ -593,26 +596,35 @@ def _assert_no_case_variant_collision(file_path: Path, checkpoint_id: Checkpoint siblings = [entry.name for entry in file_path.parent.iterdir()] except OSError as exc: # Permissions can allow lookup/creation/replacement while denying - # enumeration, so skipping the scan here would let a case-variant + # enumeration, so skipping the scan here would let a variant # save proceed unchecked. Fail closed instead. raise WorkflowCheckpointException( f"Cannot verify checkpoint ID {checkpoint_id!r} against existing checkpoints: " "the storage directory could not be listed." ) from exc + # The filesystem key is the NFC-normalized, case-folded spelling: both + # folds are lossy on the volumes that perform them, so two IDs sharing + # a key cannot coexist there. str.casefold() alone does not normalize. + + def _fs_key(name: str) -> str: + return unicodedata.normalize("NFC", name).casefold() + + expected_key = _fs_key(expected_name) variant = next( - (name for name in siblings if name != expected_name and name.casefold() == expected_name.casefold()), + (name for name in siblings if name != expected_name and _fs_key(name) == expected_key), None, ) if variant is None or expected_name in siblings: - # No case-only variant exists, or the exact spelling is already - # stored (a same-ID update on a case-sensitive volume where both + # No filesystem-folded variant exists, or the exact spelling is + # already stored (a same-ID update on a volume where distinct # spellings legitimately coexist). return raise WorkflowCheckpointException( f"Checkpoint ID {checkpoint_id!r} collides with the existing checkpoint " f"{variant.removesuffix('.json')!r}: on this filesystem both names " "resolve to one file, and saving would replace that checkpoint's data. " - "Choose a checkpoint ID that differs by more than letter case." + "Choose a checkpoint ID that differs by more than letter case or " + "Unicode normalization form." ) @classmethod @@ -636,7 +648,7 @@ async def save(self, checkpoint: WorkflowCheckpoint) -> CheckpointID: from ._checkpoint_encoding import decode_checkpoint_value, encode_checkpoint_value file_path = self._validate_file_path(checkpoint.checkpoint_id) - self._assert_no_case_variant_collision(file_path, checkpoint.checkpoint_id) + self._assert_no_variant_collision(file_path, checkpoint.checkpoint_id) checkpoint_dict = checkpoint.to_dict() # Fail at save time if encoding or restore validation fails (#8181). try: diff --git a/python/packages/core/tests/workflow/test_checkpoint_storage_conformance.py b/python/packages/core/tests/workflow/test_checkpoint_storage_conformance.py index 8c780b32624..a77f0e38671 100644 --- a/python/packages/core/tests/workflow/test_checkpoint_storage_conformance.py +++ b/python/packages/core/tests/workflow/test_checkpoint_storage_conformance.py @@ -200,3 +200,61 @@ async def test_file_save_rejects_case_variant_checkpoint_id(tmp_path: Path) -> N first.state["marker"] = "first-updated" assert await storage.save(first) == "Round-A" assert (await storage.load("Round-A")).state["marker"] == "first-updated" + + +def _filesystem_folds_unicode_normalization(path: Path) -> bool: + """True when the volume treats NFC and NFD spellings of a name as one file.""" + import unicodedata + + nfc = unicodedata.normalize("NFC", "café-norm.probe") + nfd = unicodedata.normalize("NFD", "café-norm.probe") + probe = path / nfc + probe.write_text("x") + try: + return (path / nfd).exists() + finally: + probe.unlink() + + +@pytest.mark.parametrize("mixed_case", [False, True], ids=["nfc-nfd", "case-nfc-nfd"]) +async def test_file_save_rejects_normalization_variant_checkpoint_id(tmp_path: Path, mixed_case: bool) -> None: + """An NFC/NFD spelling variant must not silently replace a checkpoint (#9466). + + APFS and HFS+ fold Unicode normalization forms, so two IDs that differ only + in normalization name one file there; str.casefold() alone does not catch + it. On volumes without that fold the two files coexist and there is nothing + to reject. + """ + import unicodedata + + if not _filesystem_folds_unicode_normalization(tmp_path): + pytest.skip("normalization-variant collision only occurs on normalization-insensitive filesystems") + first_spelling = unicodedata.normalize("NFC", "Café-1" if mixed_case else "café-1") + second_spelling = unicodedata.normalize("NFD", "café-1") + assert first_spelling != second_spelling # the test would be vacuous otherwise + + storage = FileCheckpointStorage(tmp_path / "checkpoints") + first = WorkflowCheckpoint( + workflow_name="first-workflow", + graph_signature_hash="topology-a", + checkpoint_id=first_spelling, + state={"marker": "first"}, + ) + second = WorkflowCheckpoint( + workflow_name="second-workflow", + graph_signature_hash="topology-b", + checkpoint_id=second_spelling, + state={"marker": "second"}, + ) + + first_id = await storage.save(first) + with pytest.raises(WorkflowCheckpointException, match="collides"): + await storage.save(second) + + restored = await storage.load(first_id) + assert restored.checkpoint_id == first_spelling + assert restored.state["marker"] == "first" + + # Re-saving the exact same spelling remains a normal update. + first.state["marker"] = "first-updated" + assert await storage.save(first) == first_spelling From 76b34faa507b123c407ed7070a4588c6d6cbe886 Mon Sep 17 00:00:00 2001 From: Yufeng He <40085740+he-yufeng@users.noreply.github.com> Date: Fri, 23 Oct 2026 12:07:31 +0800 Subject: [PATCH 4/4] Python: fix(core): deny case-fold aliases in load/delete and serialize variant saves --- .../agent_framework/_workflows/_checkpoint.py | 89 ++++++++++++++++--- .../core/tests/workflow/test_checkpoint.py | 23 +++-- .../test_checkpoint_storage_conformance.py | 82 +++++++++++++++++ 3 files changed, 176 insertions(+), 18 deletions(-) diff --git a/python/packages/core/agent_framework/_workflows/_checkpoint.py b/python/packages/core/agent_framework/_workflows/_checkpoint.py index 6392e791b10..0f89e2cf165 100644 --- a/python/packages/core/agent_framework/_workflows/_checkpoint.py +++ b/python/packages/core/agent_framework/_workflows/_checkpoint.py @@ -288,7 +288,7 @@ async def list_checkpoint_ids(self, *, workflow_name: str) -> list[CheckpointID] # checkpoint ID from different processes still rely on ``os.replace`` being atomic on # the underlying filesystem for the file not to be seen half-written; which write # survives is undefined. -_destination_queues: dict[Path, _DestinationQueue] = {} +_destination_queues: dict[tuple[Path, str], _DestinationQueue] = {} # A completed predecessor invokes callbacks synchronously. If cancellation lands while # a save is entering the queue, its deferred release can therefore run re-entrantly on # the same thread before the enqueue frame has left this guard. A plain Lock deadlocks @@ -324,14 +324,37 @@ class _WriteTicket: error: BaseException | None = None +def _fs_key(file_name: str) -> str: + """The filesystem's folded spelling of a filename. + + Case-insensitive volumes (the default on Windows and macOS) merge ``a`` and + ``A``; APFS and HFS+ additionally merge Unicode normalization forms. Both + folds are lossy, so two names sharing a key cannot coexist there. + ``str.casefold()`` alone does not normalize. + """ + return unicodedata.normalize("NFC", file_name).casefold() + + +def _destination_key(file_path: Path) -> tuple[Path, str]: + """The queue key for *file_path*: one entry per filesystem-folded name. + + Variant spellings of one checkpoint (``Round-A.json`` and ``round-a.json``) + resolve to a single file on a folding volume, so their writes must share one + queue or the second replace can land first and overwrite the newer one + (#9466 review). + """ + return (file_path.parent, _fs_key(file_path.name)) + + def _enqueue_write(file_path: Path) -> _WriteTicket: """Take a place in *file_path*'s queue without waiting for it.""" completion: Future[None] = Future() + key = _destination_key(file_path) with _destination_queues_guard: - queue = _destination_queues.get(file_path) + queue = _destination_queues.get(key) if queue is None: queue = _DestinationQueue() - _destination_queues[file_path] = queue + _destination_queues[key] = queue predecessor = queue.tail queue.tail = completion queue.pending += 1 @@ -492,11 +515,11 @@ def _release_one_write(ticket: _WriteTicket) -> None: if ticket.released: return ticket.released = True - queue = _destination_queues.get(ticket.path) + queue = _destination_queues.get(_destination_key(ticket.path)) if queue is not None: queue.pending -= 1 if queue.pending <= 0: - del _destination_queues[ticket.path] + del _destination_queues[_destination_key(ticket.path)] # Outside the guard: waking the successor must not happen while holding the lock its # own release will need. if not ticket.completion.done(): @@ -605,10 +628,6 @@ def _assert_no_variant_collision(file_path: Path, checkpoint_id: CheckpointID) - # The filesystem key is the NFC-normalized, case-folded spelling: both # folds are lossy on the volumes that perform them, so two IDs sharing # a key cannot coexist there. str.casefold() alone does not normalize. - - def _fs_key(name: str) -> str: - return unicodedata.normalize("NFC", name).casefold() - expected_key = _fs_key(expected_name) variant = next( (name for name in siblings if name != expected_name and _fs_key(name) == expected_key), @@ -627,6 +646,37 @@ def _fs_key(name: str) -> str: "Unicode normalization form." ) + @staticmethod + def _assert_stored_spelling(file_path: Path, checkpoint_id: CheckpointID) -> None: + """Refuse to act on a file that only a filesystem fold resolved to. + + On a case- or normalization-folding volume, ``load("round-a")`` resolves + to the stored ``Round-A.json`` and would read another checkpoint's data; + ``delete("round-a")`` would remove it. Under the fail-closed policy a + variant ID names no checkpoint at all, so both operations must stop + where the stored spelling differs (#9466 review). + """ + expected_name = f"{checkpoint_id}.json" + try: + siblings = [entry.name for entry in file_path.parent.iterdir()] + except OSError as exc: + # Enumeration can fail where lookup succeeds; acting on a fold we + # could not inspect is exactly the aliasing this guard prevents. + raise WorkflowCheckpointException( + f"Cannot verify checkpoint ID {checkpoint_id!r} against existing checkpoints: " + "the storage directory could not be listed." + ) from exc + if expected_name in siblings: + return + variant = next((name for name in siblings if _fs_key(name) == _fs_key(expected_name)), None) + if variant is None: + return + raise WorkflowCheckpointException( + f"No checkpoint found with ID {checkpoint_id}: the stored checkpoint is spelled " + f"{variant.removesuffix('.json')!r}, and on this filesystem both names resolve to " + "one file. Refusing to read or delete another checkpoint's data." + ) + @classmethod def _delete_lock(cls, file_path: Path) -> threading.Lock: """Return the process-local deletion lock for a checkpoint file.""" @@ -690,6 +740,15 @@ def _write_atomic() -> None: # filesystem). A short, ID-independent name keeps every destination # name accepted by _validate_file_path saveable regardless of # checkpoint-ID length or filesystem limits. + # + # The collision guard is repeated here, under destination ownership, + # because the pre-enqueue check cannot see a variant sibling another + # save is about to create: two first-time variant saves both observed + # "no destination" and the second replace would win (#9466 review). + # Variant spellings share one queue, so this re-check runs after the + # predecessor has actually landed. Cross-process races stay a + # documented limitation. + self._assert_no_variant_collision(file_path, checkpoint.checkpoint_id) tmp_path: Path | None = None try: # O_CREAT | O_EXCL | O_WRONLY with an explicit 0o666 mode, so the @@ -814,9 +873,13 @@ async def load(self, checkpoint_id: CheckpointID) -> WorkflowCheckpoint: Raises: WorkflowCheckpointException: If no checkpoint with the given ID exists, - or if checkpoint decoding fails. + or if checkpoint decoding fails. Also when the ID resolves to a + differently spelled checkpoint only through a filesystem + case/normalization fold: under fail-closed, that ID names no + checkpoint and the other one's data is not read. """ file_path = self._validate_file_path(checkpoint_id) + self._assert_stored_spelling(file_path, checkpoint_id) if not file_path.exists(): raise WorkflowCheckpointException(f"No checkpoint found with ID {checkpoint_id}") @@ -884,8 +947,14 @@ async def delete(self, checkpoint_id: CheckpointID) -> bool: Returns: True if the checkpoint was successfully deleted, False if no checkpoint with the given ID exists. + + Raises: + WorkflowCheckpointException: If the ID resolves to a differently spelled + checkpoint only through a filesystem case/normalization fold, or the + storage directory cannot be listed to rule that out. """ file_path = self._validate_file_path(checkpoint_id) + self._assert_stored_spelling(file_path, checkpoint_id) file_lock = self._delete_lock(file_path) def _delete() -> bool: diff --git a/python/packages/core/tests/workflow/test_checkpoint.py b/python/packages/core/tests/workflow/test_checkpoint.py index 6e13a414f8b..db69850af17 100644 --- a/python/packages/core/tests/workflow/test_checkpoint.py +++ b/python/packages/core/tests/workflow/test_checkpoint.py @@ -2277,7 +2277,8 @@ def make(name: str) -> WorkflowCheckpoint: # queued behind the holder before cancelling, or this would be testing a # cancellation that raced the enqueue instead. registry = checkpoint_module._destination_queues # pyright: ignore[reportPrivateUsage] - canonical = (Path(temp_dir) / "shared-id.json").resolve() + # The destination queue is keyed by the filesystem-folded name (#9466). + canonical = checkpoint_module._destination_key((Path(temp_dir) / "shared-id.json").resolve()) for _ in range(400): entry = registry.get(canonical) if entry is not None and entry.pending == 3: @@ -2507,7 +2508,7 @@ def test_release_write_tolerates_an_already_dropped_entry(): # No entry was ever created for this path. checkpoint_module._release_write(ticket) # pyright: ignore[reportPrivateUsage] assert ticket.completion.done() - assert path not in checkpoint_module._destination_queues # pyright: ignore[reportPrivateUsage] + assert checkpoint_module._destination_key(path) not in checkpoint_module._destination_queues # pyright: ignore[reportPrivateUsage] def test_file_checkpoint_storage_abandoned_loop_does_not_own_destination_forever(monkeypatch, tmp_path): @@ -2675,7 +2676,8 @@ def make(name: str) -> WorkflowCheckpoint: third = asyncio.create_task(storage.save(make("third"))) registry = checkpoint_module._destination_queues # pyright: ignore[reportPrivateUsage] - canonical = (Path(temp_dir) / "shared-id.json").resolve() + # The destination queue is keyed by the filesystem-folded name (#9466). + canonical = checkpoint_module._destination_key((Path(temp_dir) / "shared-id.json").resolve()) for _ in range(400): entry = registry.get(canonical) if entry is not None and entry.pending == 3: @@ -2734,7 +2736,8 @@ def gated_replace(src, dst): # noqa: ANN001, ANN202 - test shim monkeypatch.setattr(checkpoint_module.os, "replace", gated_replace) - canonical = (tmp_path / "shared-id.json").resolve() + # The destination queue is keyed by the filesystem-folded name (#9466). + canonical = checkpoint_module._destination_key((tmp_path / "shared-id.json").resolve()) observed: dict[str, bool] = {} async def main() -> None: @@ -2790,7 +2793,8 @@ def test_file_checkpoint_storage_shutdown_before_the_write_starts_does_not_hang( from agent_framework._workflows import _checkpoint as checkpoint_module - canonical = (tmp_path / "shared-id.json").resolve() + # The destination queue is keyed by the filesystem-folded name (#9466). + canonical = checkpoint_module._destination_key((tmp_path / "shared-id.json").resolve()) outcome: dict[str, object] = {} # Hold the submitted write at its first syscall. Without this the count below races the @@ -3075,7 +3079,8 @@ def gated_replace(src, dst): # noqa: ANN001, ANN202 - test shim def make(name: str) -> WorkflowCheckpoint: return WorkflowCheckpoint(workflow_name=name, graph_signature_hash="test-hash", checkpoint_id="shared-id") - canonical = (tmp_path / "shared-id.json").resolve() + # The destination queue is keyed by the filesystem-folded name (#9466). + canonical = checkpoint_module._destination_key((tmp_path / "shared-id.json").resolve()) registry = checkpoint_module._destination_queues # pyright: ignore[reportPrivateUsage] holder_loop = asyncio.new_event_loop() @@ -3172,7 +3177,8 @@ def gated_replace(src, dst): # noqa: ANN001, ANN202 - test shim def make(name: str) -> WorkflowCheckpoint: return WorkflowCheckpoint(workflow_name=name, graph_signature_hash="test-hash", checkpoint_id="shared-id") - canonical = (tmp_path / "shared-id.json").resolve() + # The destination queue is keyed by the filesystem-folded name (#9466). + canonical = checkpoint_module._destination_key((tmp_path / "shared-id.json").resolve()) registry = checkpoint_module._destination_queues # pyright: ignore[reportPrivateUsage] holder_loop = asyncio.new_event_loop() @@ -3265,7 +3271,8 @@ def gated_replace(src, dst): # noqa: ANN001, ANN202 - test shim def make(name: str) -> WorkflowCheckpoint: return WorkflowCheckpoint(workflow_name=name, graph_signature_hash="test-hash", checkpoint_id="shared-id") - canonical = (tmp_path / "shared-id.json").resolve() + # The destination queue is keyed by the filesystem-folded name (#9466). + canonical = checkpoint_module._destination_key((tmp_path / "shared-id.json").resolve()) registry = checkpoint_module._destination_queues # pyright: ignore[reportPrivateUsage] def wait_for_pending(count: int, timeout: float = 20.0) -> None: diff --git a/python/packages/core/tests/workflow/test_checkpoint_storage_conformance.py b/python/packages/core/tests/workflow/test_checkpoint_storage_conformance.py index a77f0e38671..866ee04069a 100644 --- a/python/packages/core/tests/workflow/test_checkpoint_storage_conformance.py +++ b/python/packages/core/tests/workflow/test_checkpoint_storage_conformance.py @@ -4,6 +4,7 @@ from __future__ import annotations +import asyncio from pathlib import Path from typing import Any, cast @@ -202,6 +203,87 @@ async def test_file_save_rejects_case_variant_checkpoint_id(tmp_path: Path) -> N assert (await storage.load("Round-A")).state["marker"] == "first-updated" + +async def test_file_load_rejects_case_variant_checkpoint_id(tmp_path: Path) -> None: + """A variant ID must not read another checkpoint's data (#9466 review). + + On a folding filesystem ``load("round-a")`` resolves to the stored + ``Round-A.json``; under fail-closed it names no checkpoint at all. + """ + if not _filesystem_is_case_insensitive(tmp_path): + pytest.skip("case-variant collision only occurs on case-insensitive filesystems") + storage = FileCheckpointStorage(tmp_path / "checkpoints") + first = WorkflowCheckpoint( + workflow_name="first-workflow", + graph_signature_hash="topology-a", + checkpoint_id="Round-A", + state={"marker": "first"}, + ) + await storage.save(first) + + with pytest.raises(WorkflowCheckpointException, match="spelled 'Round-A'"): + await storage.load("round-a") + + # The stored checkpoint is untouched and still reachable under its own ID. + restored = await storage.load("Round-A") + assert restored.state["marker"] == "first" + + +async def test_file_delete_rejects_case_variant_checkpoint_id(tmp_path: Path) -> None: + """A variant ID must not delete another checkpoint (#9466 review).""" + if not _filesystem_is_case_insensitive(tmp_path): + pytest.skip("case-variant collision only occurs on case-insensitive filesystems") + storage = FileCheckpointStorage(tmp_path / "checkpoints") + first = WorkflowCheckpoint( + workflow_name="first-workflow", + graph_signature_hash="topology-a", + checkpoint_id="Round-A", + state={"marker": "first"}, + ) + await storage.save(first) + + with pytest.raises(WorkflowCheckpointException, match="spelled 'Round-A'"): + await storage.delete("round-a") + + assert (await storage.load("Round-A")).state["marker"] == "first" + assert await storage.list_checkpoint_ids(workflow_name="first-workflow") == ["Round-A"] + + +async def test_concurrent_variant_saves_do_not_overwrite(tmp_path: Path) -> None: + """Two first-time variant saves must not both pass the guard (#9466 review). + + The collision check used to run before the write was queued, so both saves + observed "no destination" and the second atomic replace silently won. + """ + if not _filesystem_is_case_insensitive(tmp_path): + pytest.skip("case-variant collision only occurs on case-insensitive filesystems") + storage = FileCheckpointStorage(tmp_path / "checkpoints") + upper = WorkflowCheckpoint( + workflow_name="upper-workflow", + graph_signature_hash="topology-a", + checkpoint_id="Round-B", + state={"marker": "upper"}, + ) + lower = WorkflowCheckpoint( + workflow_name="lower-workflow", + graph_signature_hash="topology-b", + checkpoint_id="round-b", + state={"marker": "lower"}, + ) + + results = await asyncio.gather(storage.save(upper), storage.save(lower), return_exceptions=True) + errors = [r for r in results if isinstance(r, WorkflowCheckpointException)] + saved_ids = [r for r in results if isinstance(r, str)] + assert len(errors) == 1, f"expected exactly one save to be rejected, got {results!r}" + assert "collides" in str(errors[0]) + assert len(saved_ids) == 1 + + # Whichever spelling won keeps its own data; nothing is silently replaced. + stored = await storage.list_checkpoint_ids(workflow_name=f"{'upper' if saved_ids[0] == 'Round-B' else 'lower'}-workflow") + assert stored == [saved_ids[0]] + assert (await storage.load(saved_ids[0])).state["marker"] == ("upper" if saved_ids[0] == "Round-B" else "lower") + + def _filesystem_folds_unicode_normalization(path: Path) -> bool: """True when the volume treats NFC and NFD spellings of a name as one file.""" import unicodedata