Add snapshot file locking

.gitignore

@@ -15,3 +15,5 @@ htmlcov/
 .mypy_cache/
 .ruff_cache/
 .pyright/
+
+.todo/

README.md

@@ -80,6 +80,49 @@ article = Article(
 loaded = Article.snapshots.get("dusk-court")
 ```
 
+## Coordinating Shared Files
+
+When two local processes may update the same file, wrap the short
+read-modify-save section in `snapshot.locked(reload=True)`. The lock is
+cooperative and local to the machine, using a `.lock` file beside the snapshot.
+Snapshot filenames ending in `.lock` are reserved for these lock sidecars.
+
+```python
+from snapclass import snapclass, Stash, Fresh
+
+
+@snapclass("{self.name}.yml", stash=Stash("./runs"), manual=True, require_lock=True)
+class WorkflowState:
+    name: str
+    steps: list[str] = Fresh.List
+
+
+state = WorkflowState("daily-run")
+
+with state.snapshot.locked(reload=True):
+    state.steps.append("started")
+    state.save()
+```
+
+For async workflows, keep the locked block short. Do the slow work after the
+save has released the file lock:
+
+```python
+with state.snapshot.locked(reload=True):
+    state.steps.append("started")
+    state.save()
+
+await do_work()
+
+with state.snapshot.locked(reload=True):
+    state.steps.append("finished")
+    state.save()
+```
+
+`require_lock=True` is optional, but useful for manual models where every save
+should go through this pattern. It makes `state.save()` raise unless it is
+called inside `state.snapshot.locked(...)`.
+
 ## FAQ
 
 ### Why use `snapclass` over `datafiles`?

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "snapclass"
-version = "0.1.2"
+version = "0.1.3"
 description = "Human-readable file persistence for Python dataclasses."
 readme = "README.md"
 requires-python = ">=3.11"

skills/snapclass-fluency/SKILL.md

@@ -406,6 +406,55 @@ note.save()
 note.load()
 ```
 
+Use `snapshot.locked(reload=True)` when multiple local processes or workers may
+write the same snapshot file. The intended pattern is short, explicit, and
+file-centered:
+
+```python
+from snapclass import snapclass, Stash, Fresh
+
+
+@snapclass("{self.name}.yml", stash=Stash("./runs"), manual=True, require_lock=True)
+class WorkflowState:
+    name: str
+    steps: list[str] = Fresh.List
+
+
+state = WorkflowState("daily-run")
+
+with state.snapshot.locked(reload=True):
+    state.steps.append("started")
+    state.save()
+```
+
+`locked(reload=True)` acquires a cooperative per-file OS lock, reloads the
+latest file contents while the lock is held, lets the caller mutate the object,
+and expects an explicit save before leaving the block. Use `require_lock=True`
+for shared persisted models where saving outside `snapshot.locked(...)` should
+be an error. `require_lock=True` belongs with `manual=True`.
+
+In async workflows, still use the synchronous context manager and keep the block
+tiny. Do the slow awaitable work after releasing the lock:
+
+```python
+with state.snapshot.locked(reload=True):
+    state.steps.append("started")
+    state.save()
+
+await do_work()
+
+with state.snapshot.locked(reload=True):
+    state.steps.append("finished")
+    state.save()
+```
+
+The lock is local-machine, cross-process coordination for cooperative snapclass
+writers. It is a good fit for two Python backends sharing the same ordinary local
+file. Raw writers that ignore the `.lock` side file can still race, and
+network/cloud-synced filesystems, containers, mounted volumes, and mixed
+WSL/Windows access need explicit validation before relying on the lock.
+Snapshot filenames ending in `.lock` are reserved for snapclass lock sidecars.
+
 `snapshot.data` is the serialized mapping before file formatting. `snapshot.text` is the formatted file text for the current pattern or formatter. Setting `snapshot.text` writes the file directly, reloads the object, and still honors conflict policy.
 
 Patternless `Model` or `create_model(...)` objects can use `.snapshot.data` and `.snapshot.text` as projections, but saving requires a pattern.
@@ -962,6 +1011,7 @@ class Prompt(Model):
         snapshot_pattern = "{self.name}.yml"
         snapshot_stash = Stash("./prompts")
         snapshot_manual = True
+        snapshot_require_lock = False
         snapshot_defaults = False
         snapshot_infer = False
         snapshot_fields = None

src/snapclass/_locks.py

@@ -0,0 +1,142 @@
+from __future__ import annotations
+
+import errno
+import os
+import threading
+import time
+from collections.abc import Iterator
+from contextlib import contextmanager
+from pathlib import Path
+from typing import BinaryIO
+
+
+class _PathLockState:
+    def __init__(self) -> None:
+        self.lock = threading.RLock()
+        self.depth = 0
+        self.handle: BinaryIO | None = None
+
+
+_LOCKS: dict[Path, _PathLockState] = {}
+_LOCKS_GUARD = threading.Lock()
+
+
+def write_lock_for(path: Path) -> threading.RLock:
+    return _lock_state_for(path).lock
+
+
+@contextmanager
+def locked_path(path: Path) -> Iterator[None]:
+    normalized_path = _normalized_path(path)
+    state = _lock_state_for(normalized_path)
+    state.lock.acquire()
+    try:
+        if state.depth == 0:
+            state.handle = _acquire_os_lock(normalized_path)
+        state.depth += 1
+        try:
+            yield
+        finally:
+            state.depth -= 1
+            if state.depth == 0:
+                handle = state.handle
+                state.handle = None
+                if handle is not None:
+                    _release_os_lock(handle)
+    finally:
+        state.lock.release()
+
+
+def _lock_state_for(path: Path) -> _PathLockState:
+    key = _normalized_path(path)
+    with _LOCKS_GUARD:
+        state = _LOCKS.get(key)
+        if state is None:
+            state = _PathLockState()
+            _LOCKS[key] = state
+        return state
+
+
+def _normalized_path(path: Path) -> Path:
+    absolute = path if path.is_absolute() else Path.cwd() / path
+    # Atomic replace writes to the final path itself, so resolve parent
+    # directories but keep the leaf name instead of following a leaf symlink.
+    return absolute.parent.resolve(strict=False) / absolute.name
+
+
+def _lock_path_for(path: Path) -> Path:
+    return path.with_name(f"{path.name}.lock")
+
+
+def _is_lock_path(path: Path) -> bool:
+    return path.name.lower().endswith(".lock")
+
+
+def _acquire_os_lock(path: Path) -> BinaryIO:
+    lock_path = _lock_path_for(path)
+    lock_path.parent.mkdir(parents=True, exist_ok=True)
+    handle = lock_path.open("a+b")
+    try:
+        if os.name == "nt":
+            _acquire_windows_lock(handle)
+        else:
+            _acquire_posix_lock(handle)
+    except Exception:
+        handle.close()
+        raise
+    return handle
+
+
+def _release_os_lock(handle: BinaryIO) -> None:
+    try:
+        if os.name == "nt":
+            _release_windows_lock(handle)
+        else:
+            _release_posix_lock(handle)
+    finally:
+        handle.close()
+
+
+def _acquire_windows_lock(handle: BinaryIO) -> None:
+    import msvcrt
+
+    handle.seek(0, os.SEEK_END)
+    if handle.tell() == 0:
+        handle.write(b"\0")
+        handle.flush()
+    while True:
+        try:
+            handle.seek(0)
+            msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1)
+            return
+        except OSError as exc:
+            if not _is_windows_lock_contention(exc):
+                raise
+            time.sleep(0.05)
+
+
+def _is_windows_lock_contention(exc: OSError) -> bool:
+    winerror = getattr(exc, "winerror", None)
+    if winerror is not None:
+        return winerror in {32, 33}
+    # CPython's msvcrt.locking reports byte-range lock contention this way.
+    return exc.errno in {errno.EACCES, errno.EDEADLK}
+
+
+def _release_windows_lock(handle: BinaryIO) -> None:
+    import msvcrt
+
+    handle.seek(0)
+    msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
+
+
+def _acquire_posix_lock(handle: BinaryIO) -> None:
+    import fcntl
+
+    fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
+
+
+def _release_posix_lock(handle: BinaryIO) -> None:
+    import fcntl
+
+    fcntl.flock(handle.fileno(), fcntl.LOCK_UN)

src/snapclass/collections.py

@@ -49,7 +49,12 @@ def get_or_create(self, *args: Any, **kwargs: Any) -> Any:
         instance = self._empty_instance(*args, **kwargs, include_defaults=True)
         _attach_snapshot(instance, self.model.__snapclass_config__, self._stash)
         initial_path = instance.snapshot._require_path()
-        with _write_lock_for(initial_path):
+        lock = (
+            instance.snapshot.locked()
+            if getattr(instance.snapshot, "require_lock", False)
+            else _write_lock_for(initial_path)
+        )
+        with lock:
             if instance.snapshot.exists:
                 instance.snapshot.load(_initial=True)
             else: