fix(task-21): registry identity is the file (st_dev/st_ino), not its spelling — a hard link, junction or case variant reloads instead of registering twice

`_file_identity` keyed on normcase(abspath), which folds case and separators
only, and only on Windows. Two names for one file — a hard link (which
Path.resolve() cannot fold either), an NTFS junction reached in-process, or
a case variant on macOS's case-insensitive APFS where normcase is the
identity — registered two entries: both rows said `saved: True`, a save from
either overwrote the other's bytes, and closing the 'clean' one dropped its
work as `discarded_changes: False`.

Identity now asks the disk: `os.stat` (st_dev, st_ino) when the path exists
(the samefile identity), `normcase(abspath)` only for a path not on disk yet
(a save target, or an owner whose file was removed) or a filesystem that
reports no inode. `_register_document`, `_path_holder`,
`_resolve_document_key`, the `_doc_path` setter and the quarantine's
same-path warning all go through the one helper. An untitled entry holds no
file and never matches a path.

Tests: the `_case_folding_fs` guard probes the temp filesystem instead of
`os.path.normcase`, so the case-variant tests run on macOS where they used
to skip while the bug was live. New: hard link (backend and over the wire),
junction (mklink /J; a directory symlink elsewhere), a case variant with
normcase given its POSIX definition, and the spelling fallback for a save
target that is not on disk. Four of the five fail on the previous backend.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd
This commit is contained in:
Umut CAN
2026-09-18 10:08:12 +03:00
co-authored by Claude Opus 5
parent 31daac022f
commit bbc9aeeb63
2 changed files with 196 additions and 27 deletions
+41 -24
View File
@@ -1109,9 +1109,9 @@ class EzdxfBackend(AutoCADBackend):
)
state = self._state
state.path = value
# Re-key only when the *file* changes: another spelling of the file
# the entry already holds (a case variant on Windows) keeps the key a
# caller may be holding.
# Re-key only when the *file* changes: another name for the file the
# entry already holds (a case variant, a hard link, a junction) keeps
# the key a caller may be holding.
if not self._same_file(new_key, self._active_key):
del self._docs[self._active_key]
state.key = new_key
@@ -1119,19 +1119,35 @@ class EzdxfBackend(AutoCADBackend):
self._active_key = new_key
@staticmethod
def _file_identity(path: str) -> str:
"""The string two spellings of one file share: ``normcase(abspath(path))``.
def _file_identity(path: str) -> tuple:
"""What two names of one file share - the *file*, not its spelling.
Every registry comparison goes through this. On Windows it folds case
and separators, so ``C:\\Work\\Gear.dxf`` and ``c:/work/gear.dxf`` are one
file; on POSIX ``normcase`` is the identity and only ``..`` / ``.`` /
duplicate separators are folded. Keys themselves stay spelled as the
caller spelled them - the identity is for comparing, not for showing.
Every registry comparison goes through this. A path that exists is
identified by ``os.stat`` ``(st_dev, st_ino)`` - the same identity
``os.path.samefile`` uses - so a case variant on a case-insensitive
volume (NTFS, default APFS), a hard link, an NTFS junction or a
symlink all name the one file, and the registry keeps one entry for
it. ``normcase(abspath(path))`` was the earlier rule and folded only
case and separators, and only on Windows: two names for one file
registered twice, both rows said ``saved``, and a save from either
overwrote the other's bytes. It stays as the fallback for a path that
is not on disk yet (a save target) or that the OS cannot stat, and for
a filesystem that reports no inode. Keys themselves stay spelled as
the caller spelled them - the identity is for comparing, not showing.
"""
try:
return os.path.normcase(os.path.abspath(path))
absolute = os.path.abspath(path)
except (OSError, ValueError): # a path this OS cannot normalise
return os.path.normcase(path)
return ("spelling", os.path.normcase(path))
try:
st = os.stat(absolute)
except (OSError, ValueError):
return ("spelling", os.path.normcase(absolute))
if st.st_ino == 0 and st.st_dev == 0:
# A filesystem without stable file identity (some network shares):
# fall back to the spelling rather than call every file one file.
return ("spelling", os.path.normcase(absolute))
return ("file", st.st_dev, st.st_ino)
@classmethod
def _same_file(cls, a: str, b: str) -> bool:
@@ -1140,17 +1156,16 @@ class EzdxfBackend(AutoCADBackend):
def _path_holder(self, path: str, *, exclude: str | None = None) -> str | None:
"""Key of the entry (other than ``exclude``) that holds ``path``, or None.
Compared by ``_file_identity`` so that on Windows two spellings of one
file count as the same file; on POSIX that is plain equality.
Compared by ``_file_identity``: the entry's path and its key are both
names of the file it holds, so either matching ``path`` - by inode
when both are on disk, by ``normcase`` spelling otherwise - means the
file is held. An untitled entry holds no file and never matches.
"""
wanted = self._file_identity(path)
for key, state in self._docs.items():
if key == exclude:
if key == exclude or not state.path:
continue
candidates = [key]
if state.path:
candidates.append(state.path)
if any(self._file_identity(c) == wanted for c in candidates):
if any(self._file_identity(c) == wanted for c in (key, state.path)):
return key
return None
@@ -1318,8 +1333,9 @@ class EzdxfBackend(AutoCADBackend):
its place — the pre-registry behaviour of ``drawing_open``, kept so a
save-then-reopen verification still reads the disk. "Already open" is
decided by ``_path_holder`` (file identity), not by the key string:
the same file spelled with another case used to register a second
entry, after which every lookup of either was refused as ambiguous.
the same file under another name (a case variant, a hard link, a
junction) used to register a second entry, after which a save from
either silently overwrote the other's bytes.
"""
if path is None:
self._untitled_counter += 1
@@ -1365,9 +1381,10 @@ class EzdxfBackend(AutoCADBackend):
# say so rather than hand back whichever entry the key names.
self._refuse_shared_path(wanted, self._docs[wanted])
return wanted
# Paths compare by file identity (``normcase`` on Windows), as the
# live engine's ``_com_find_document`` lowercases before comparing: a
# lowercase drive letter or a case variant names the same document.
# Paths compare by file identity (the inode when the file is on disk),
# as the live engine's ``_com_find_document`` resolves before comparing:
# a lowercase drive letter, a case variant, a hard link or a junction
# names the same document.
as_path = self._file_identity(wanted)
by_path = [
key
+155 -3
View File
@@ -20,6 +20,7 @@ it; the strict path is the new tool.
from __future__ import annotations
import os
import tempfile
import types
import pytest
@@ -213,13 +214,56 @@ async def test_resolver_refuses_a_shared_path_even_on_an_exact_key_hit(backend,
assert len(await backend.document_list()) == 2, "a refusal closes nothing"
# Two spellings of one file only exist where the filesystem folds case;
# ``os.path.normcase`` is the identity on POSIX, so these tests skip there.
def _filesystem_folds_case() -> bool:
"""Whether the temp filesystem answers to a case variant of a file's name.
Probed on disk rather than read off ``os.path.normcase``: the platform's
``normcase`` is the identity on macOS while the default APFS volume folds
case, so a ``normcase`` guard skipped every case-variant test exactly
where the old spelling-based registry identity was wrong.
"""
with tempfile.TemporaryDirectory() as folder:
probe = os.path.join(folder, "Probe.dxf")
with open(probe, "w", encoding="utf-8"):
pass
return os.path.exists(os.path.join(folder, "probe.dxf"))
# Two spellings of one file only exist where the filesystem folds case.
_case_folding_fs = pytest.mark.skipif(
os.path.normcase("A") == "A", reason="filesystem is case-sensitive"
not _filesystem_folds_case(), reason="filesystem is case-sensitive"
)
def _hard_link(target: str, link: str) -> None:
try:
os.link(target, link)
except (OSError, NotImplementedError, AttributeError) as exc:
pytest.skip(f"hard links unavailable here: {exc}")
def _junction(target_dir: str, link_dir: str) -> None:
"""A second name for a directory: an NTFS junction, or a symlink elsewhere."""
if os.name == "nt":
try:
import _winapi
_winapi.CreateJunction(target_dir, link_dir)
return
except (ImportError, AttributeError, OSError) as exc:
pytest.skip(f"NTFS junctions unavailable here: {exc}")
try:
os.symlink(target_dir, link_dir, target_is_directory=True)
except (OSError, NotImplementedError) as exc:
pytest.skip(f"directory symlinks unavailable here: {exc}")
def _modelspace_types(path) -> list[str]:
import ezdxf
return [e.dxftype() for e in ezdxf.readfile(str(path)).modelspace()]
@_case_folding_fs
async def test_opening_a_case_variant_of_an_open_file_reloads_it(backend, tmp_path):
"""Registration keys by file identity, not by spelling.
@@ -278,6 +322,93 @@ async def test_saving_to_a_case_variant_of_the_own_path_keeps_one_entry(backend,
assert (await backend.document_activate(key))["ok"] is True
async def test_a_hard_link_to_an_open_file_reloads_instead_of_registering_twice(backend, tmp_path):
"""Registry identity is the file (``st_dev``/``st_ino``), not the spelling.
``normcase(abspath)`` folded case and separators only, so a hard link -
which ``Path.resolve()`` cannot fold either - registered a second entry:
both rows said ``saved: True``, a save from either overwrote the other's
bytes, and closing the 'clean' one dropped its work as ``discarded_changes:
False``.
"""
target = tmp_path / "gear.dxf"
link = tmp_path / "gear_link.dxf"
await backend.drawing_save(str(target))
_hard_link(str(target), str(link))
assert os.path.samefile(target, link)
await backend.entity_create_line(0, 0, 1, 1) # unsaved edit
reopened = await backend.drawing_open(str(link))
assert reopened["reloaded"] is True and reopened["open_documents"] == 1
rows = await backend.document_list()
assert len(rows) == 1 and rows[0]["entity_count"] == 0, "the reload reads the disk"
key = rows[0]["key"]
await backend.entity_create_circle(0, 0, 5)
await backend.drawing_save()
assert _modelspace_types(target) == ["CIRCLE"], "one entry, one file, one save"
assert (await backend.document_activate(str(target)))["active"] == key
assert (await backend.document_activate(str(link)))["active"] == key
async def test_a_junction_to_an_open_file_reloads_instead_of_registering_twice(backend, tmp_path):
"""Same rule through a directory junction (``mklink /J``), which
``os.path.abspath`` leaves untouched - the route in-process callers take."""
real_dir = tmp_path / "real"
real_dir.mkdir()
target = real_dir / "gear.dxf"
await backend.drawing_save(str(target))
_junction(str(real_dir), str(tmp_path / "via_junction"))
aliased = tmp_path / "via_junction" / "gear.dxf"
assert os.path.samefile(target, aliased)
await backend.entity_create_line(0, 0, 1, 1)
reopened = await backend.drawing_open(str(aliased))
assert reopened["reloaded"] is True and reopened["open_documents"] == 1
await backend.entity_create_circle(0, 0, 5)
await backend.drawing_save()
assert _modelspace_types(target) == ["CIRCLE"]
await backend.drawing_new()
with pytest.raises(ValueError, match="already open as document"):
await backend.drawing_save_as(str(aliased))
assert _modelspace_types(target) == ["CIRCLE"], "a refusal writes nothing"
closed = await backend.document_close(str(aliased))
assert closed["discarded_changes"] is False and closed["open_documents"] == 1
@_case_folding_fs
async def test_case_variant_identity_does_not_depend_on_normcase(backend, tmp_path, monkeypatch):
"""macOS: ``os.path.normcase`` is the identity, the default APFS volume
folds case. Reproduced here by giving ``normcase`` its POSIX definition -
the registry must still see one file, because it asks the disk."""
monkeypatch.setattr(os.path, "normcase", lambda s: s)
path = tmp_path / "Gear.dxf"
await backend.drawing_save(str(path))
await backend.entity_create_line(0, 0, 1, 1)
reopened = await backend.drawing_open(str(path).lower())
assert reopened["reloaded"] is True and reopened["open_documents"] == 1
await backend.entity_create_circle(0, 0, 5)
await backend.drawing_save()
assert _modelspace_types(path) == ["CIRCLE"]
assert len(await backend.document_list()) == 1
async def test_save_target_that_does_not_exist_yet_is_compared_by_spelling(backend, tmp_path):
"""A path with no file behind it cannot be stat'ed; it is still one file
with the spelling that another entry holds (an owner whose file was
removed from disk), and a save onto it is refused."""
path = tmp_path / "gone.dxf"
await backend.drawing_save(str(path))
path.unlink()
await backend.drawing_new()
with pytest.raises(ValueError, match="already open as document"):
await backend.drawing_save_as(str(path))
assert not path.exists(), "a refusal writes nothing"
assert (await backend.drawing_save_as(str(tmp_path / "other.dxf")))["ok"] is True
async def test_drawing_close_with_save_refuses_a_shared_path(backend, tmp_path):
"""``drawing_close`` bypasses the resolver, so the close itself must check.
@@ -532,6 +663,27 @@ async def test_document_activate_accepts_a_lowercase_drive_over_the_wire(monkeyp
assert closed["closed"] == os.path.abspath(spelled) and closed["open_documents"] == 1
async def test_hard_link_over_the_wire_is_one_document(monkeypatch, tmp_path):
"""Route (a) of the review finding: ``validate_path().resolve()`` does not
fold a hard link, so the wire hands the backend a second name for the
open file. One entry, one save, and the disk holds the last edit."""
monkeypatch.setenv("AUTOCAD_MCP_BACKEND", "ezdxf")
target = tmp_path / "gear.dxf"
link = tmp_path / "gear_link.dxf"
async with Client(server.mcp) as client:
await client.call_tool("drawing_new", {"bootstrap": False})
await client.call_tool("drawing_save", {"path": str(target)})
_hard_link(str(target), str(link))
await client.call_tool("entity_create_line", {"x1": 0, "y1": 0, "x2": 1, "y2": 1})
opened = (await client.call_tool("drawing_open", {"path": str(link)})).structured_content
assert opened["reloaded"] is True and opened["open_documents"] == 1
await client.call_tool("entity_create_circle", {"cx": 0, "cy": 0, "radius": 5})
await client.call_tool("drawing_save", {})
rows = (await client.call_tool("document_list", {})).structured_content
assert rows["count"] == 1 and rows["documents"][0]["saved"] is True
assert _modelspace_types(target) == ["CIRCLE"]
# ── live engine, against a fake ActiveX surface ──────────────────────────────