diff --git a/.gitignore b/.gitignore
index 95bd42f..023b37a 100644
--- a/.gitignore
+++ b/.gitignore
@@ -15,3 +15,5 @@ htmlcov/
 .mypy_cache/
 .ruff_cache/
 .pyright/
+
+.todo/
diff --git a/README.md b/README.md
index c6cb51d..1dfd35b 100644
--- a/README.md
+++ b/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`?
diff --git a/pyproject.toml b/pyproject.toml
index 0d6e9c1..fd023d0 100644
--- a/pyproject.toml
+++ b/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"
diff --git a/skills/snapclass-fluency/SKILL.md b/skills/snapclass-fluency/SKILL.md
index d7b7f41..367b7f5 100644
--- a/skills/snapclass-fluency/SKILL.md
+++ b/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
diff --git a/src/snapclass/_locks.py b/src/snapclass/_locks.py
new file mode 100644
index 0000000..13d7bcd
--- /dev/null
+++ b/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)
diff --git a/src/snapclass/collections.py b/src/snapclass/collections.py
index 8ae7145..b075a1d 100644
--- a/src/snapclass/collections.py
+++ b/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:
diff --git a/src/snapclass/schemas.py b/src/snapclass/schemas.py
index 357d988..c49f6d0 100644
--- a/src/snapclass/schemas.py
+++ b/src/snapclass/schemas.py
@@ -9,7 +9,6 @@
 import re
 import sys
 import tempfile
-import threading
 import time
 import types
 import warnings
@@ -20,6 +19,11 @@
 from typing import Any, Callable, Union, get_args, get_origin, get_type_hints, is_typeddict
 
 from . import formatters, serializers, sessions, sidecar
+from ._locks import (
+    _is_lock_path,
+    locked_path as _locked_path,
+    write_lock_for as _shared_write_lock_for,
+)
 from .collections import Collection, CollectionDescriptor
 from .paths import safe_path_placeholder
 from .formatters import FileFormatter
@@ -32,8 +36,6 @@
 _INFERRED_HINTS_ATTR = "__snapclass_inferred_hints__"
 _DEFAULT_CACHE_ATTR = "__snapclass_default_cache__"
 _PENDING_SIDECARS_ATTR = "__snapclass_pending_sidecars__"
-_WRITE_LOCKS: dict[Path, threading.RLock] = {}
-_WRITE_LOCKS_GUARD = threading.Lock()
 
 
 class SnapclassError(Exception):
@@ -68,6 +70,7 @@ class Config:
     extras_field: str | None = None
     migrate: Callable[..., Mapping[str, Any] | None] | None = None
     conflict: str = "overwrite"
+    require_lock: bool = False
     type_hints: dict[str, Any] = dataclasses.field(default_factory=dict)
 
 
@@ -86,6 +89,7 @@ class Meta:
     snapshot_extras_field: str | None = None
     snapshot_migrate: Callable[..., Mapping[str, Any] | None] | None = None
     snapshot_conflict: str = "overwrite"
+    snapshot_require_lock: bool = False
 
 
 def snapclass(
@@ -104,12 +108,17 @@ def snapclass(
     extras_field: str | None = None,
     migrate: Callable[..., Mapping[str, Any] | None] | None = None,
     conflict: str = "overwrite",
+    require_lock: bool = False,
     **dataclass_kwargs: Any,
 ):
     if pattern is None:
+        if require_lock:
+            raise ValueError("require_lock=True requires a persisted snapshot pattern")
         return dataclasses.dataclass(**dataclass_kwargs)
 
     if callable(pattern):
+        if require_lock:
+            raise ValueError("require_lock=True requires a persisted snapshot pattern")
         return dataclasses.dataclass(pattern)
 
     def decorate(cls: type):
@@ -117,6 +126,8 @@ def decorate(cls: type):
             cls = _dataclass_with_sidecars(cls, **dataclass_kwargs)
         unknown_policy = _normalize_unknown_policy(unknown, extras_field)
         conflict_policy = _normalize_conflict_policy(conflict)
+        _validate_snapshot_pattern(pattern)
+        _validate_require_lock(pattern, manual, require_lock)
         _validate_extras_field(cls, unknown_policy, extras_field)
         config = Config(
             pattern=pattern,
@@ -133,6 +144,7 @@ def decorate(cls: type):
             extras_field=extras_field,
             migrate=migrate,
             conflict=conflict_policy,
+            require_lock=require_lock,
         )
         config.type_hints = _resolve_type_hints(cls)
         _install(cls, config)
@@ -153,6 +165,7 @@ def create_model(
     write_delay: float | None = None,
     migrate: Callable[..., Mapping[str, Any] | None] | None = None,
     conflict: str | None = None,
+    require_lock: bool | None = None,
 ) -> type:
     if not dataclasses.is_dataclass(cls):
         raise ValueError(f"{cls} must be a dataclass")
@@ -193,6 +206,9 @@ def create_model(
     resolved_conflict = conflict if conflict is not None else (
         getattr(meta, "snapshot_conflict", "overwrite") if meta is not None else "overwrite"
     )
+    resolved_require_lock = require_lock if require_lock is not None else (
+        getattr(meta, "snapshot_require_lock", False) if meta is not None else False
+    )
     _install_model_config(
         cls,
         pattern=resolved_pattern,
@@ -208,6 +224,7 @@ def create_model(
         extras_field=extras_field,
         migrate=resolved_migrate,
         conflict=resolved_conflict,
+        require_lock=resolved_require_lock,
     )
     cls.Meta = Meta(
         snapshot_fields=resolved_fields,
@@ -223,6 +240,7 @@ def create_model(
         snapshot_extras_field=extras_field,
         snapshot_migrate=resolved_migrate,
         snapshot_conflict=_normalize_conflict_policy(resolved_conflict),
+        snapshot_require_lock=resolved_require_lock,
     )
     return cls
 
@@ -243,15 +261,19 @@ def _install_model_config(
     extras_field: str | None = None,
     migrate: Callable[..., Mapping[str, Any] | None] | None = None,
     conflict: str = "overwrite",
+    require_lock: bool = False,
 ) -> None:
     unknown_policy = _normalize_unknown_policy(unknown, extras_field)
     conflict_policy = _normalize_conflict_policy(conflict)
+    resolved_manual = True if pattern is None else manual
+    _validate_snapshot_pattern(pattern)
+    _validate_require_lock(pattern, resolved_manual, require_lock)
     _validate_extras_field(cls, unknown_policy, extras_field)
     config = Config(
         pattern=pattern,
         stash=stash,
         module_dir=_module_dir_for(cls),
-        manual=True if pattern is None else manual,
+        manual=resolved_manual,
         defaults=defaults,
         infer=infer,
         fields=fields,
@@ -262,6 +284,7 @@ def _install_model_config(
         extras_field=extras_field,
         migrate=migrate,
         conflict=conflict_policy,
+        require_lock=require_lock,
     )
     config.type_hints = _resolve_type_hints(cls)
     _install(cls, config)
@@ -311,6 +334,7 @@ def __init_subclass__(cls, **kwargs: Any) -> None:
             extras_field=getattr(meta, "snapshot_extras_field", None),
             migrate=getattr(meta, "snapshot_migrate", None),
             conflict=getattr(meta, "snapshot_conflict", "overwrite"),
+            require_lock=getattr(meta, "snapshot_require_lock", False),
         )
 
 
@@ -330,10 +354,13 @@ def sync(
     extras_field: str | None = None,
     migrate: Callable[..., Mapping[str, Any] | None] | None = None,
     conflict: str = "overwrite",
+    require_lock: bool = False,
 ) -> object:
     cls = instance.__class__
     unknown_policy = _normalize_unknown_policy(unknown, extras_field)
     conflict_policy = _normalize_conflict_policy(conflict)
+    _validate_snapshot_pattern(pattern)
+    _validate_require_lock(pattern, manual, require_lock)
     _validate_extras_field(cls, unknown_policy, extras_field)
     config = Config(
         pattern=pattern,
@@ -350,6 +377,7 @@ def sync(
         extras_field=extras_field,
         migrate=migrate,
         conflict=conflict_policy,
+        require_lock=require_lock,
     )
     config.type_hints = _safe_type_hints(cls)
     if not hasattr(cls, "__snapclass_config__"):
@@ -704,6 +732,7 @@ def __init__(
         infer: bool | None = None,
         minimal_diffs: bool | None = None,
         write_delay: float | None = None,
+        require_lock: bool | None = None,
         root: "Snapshot | None" = None,
     ) -> None:
         self._instance = instance
@@ -717,6 +746,7 @@ def __init__(
                 fields=fields or {},
                 minimal_diffs=minimal_diffs,
                 write_delay=write_delay,
+                require_lock=False if require_lock is None else require_lock,
             )
             config.type_hints = _safe_type_hints(instance.__class__)
         self._config = config
@@ -728,6 +758,8 @@ def __init__(
         self._loaded_data: dict[str, Any] | None = None
         self._loaded_path: Path | None = None
         self._ready = False
+        self._lock_depth = 0
+        self._locked_path: Path | None = None
 
     @property
     def classname(self) -> str:
@@ -753,27 +785,39 @@ def _pattern(self, value: str | None) -> None:
     @property
     def path(self) -> Path | None:
         if self._path_override is not None:
+            _validate_snapshot_path(self._path_override)
             return self._path_override
         if not self._config.pattern:
             return None
         formatted = self._config.pattern.format(self=_FormatProxy(self._instance))
         path = Path(formatted)
         if path.is_absolute():
+            _validate_snapshot_path(path)
             return path
         if _is_home_relative(path):
-            return path.expanduser().resolve()
+            resolved = path.expanduser().resolve()
+            _validate_snapshot_path(resolved)
+            return resolved
         stash = self._stash or self._config.stash
         _reject_relative_traversal(path, "snapshot pattern")
         if stash is not None:
-            return stash.path / path
+            resolved = stash.path / path
+            _validate_snapshot_path(resolved)
+            return resolved
         if self._config.pattern.startswith("./"):
-            return path.resolve()
+            resolved = path.resolve()
+            _validate_snapshot_path(resolved)
+            return resolved
         root = self._config.module_dir or Path.cwd()
-        return (root / path).resolve()
+        resolved = (root / path).resolve()
+        _validate_snapshot_path(resolved)
+        return resolved
 
     @path.setter
     def path(self, value: str | os.PathLike[str]) -> None:
-        self._path_override = Path(value)
+        path = Path(value)
+        _validate_snapshot_path(path)
+        self._path_override = path
 
     @property
     def relpath(self) -> Path | None:
@@ -821,6 +865,12 @@ def infer(self) -> bool:
             return self._root.infer
         return self._config.infer
 
+    @property
+    def require_lock(self) -> bool:
+        if self._root is not None:
+            return self._root.require_lock
+        return self._config.require_lock
+
     @property
     def stash(self) -> Stash | None:
         return self._stash or self._config.stash
@@ -851,6 +901,7 @@ def text(self, value: str) -> None:
         __tracebackhide__ = sessions.HIDDEN_TRACEBACK
         path = self._require_path()
         with _write_lock_for(path):
+            self._check_required_lock(path)
             self._check_write_conflict(path)
             _write_text_atomic(
                 path,
@@ -870,6 +921,7 @@ def save(
             self.path = path
         current_path = self._require_path()
         with _write_lock_for(current_path):
+            self._check_required_lock(current_path)
             self._check_write_conflict(current_path)
             sidecar.reconcile_before_save(self._instance, current_path)
             data = _to_data(
@@ -922,6 +974,27 @@ def load(
         finally:
             object.__setattr__(self._instance, "_snapclass_loading", False)
 
+    @contextmanager
+    def locked(self, *, reload: bool = False) -> Iterator["Snapshot"]:
+        __tracebackhide__ = sessions.HIDDEN_TRACEBACK
+        current_path = self._require_path()
+        if self._lock_depth and self._locked_path != current_path:
+            raise SnapclassError(
+                "Snapshot path changed while locked; keep path fields stable inside "
+                "snapshot.locked()"
+            )
+        with _locked_path(current_path):
+            previous_path = self._locked_path
+            self._locked_path = current_path
+            self._lock_depth += 1
+            try:
+                if reload and current_path.exists():
+                    self.load()
+                yield self
+            finally:
+                self._lock_depth -= 1
+                self._locked_path = previous_path if self._lock_depth else None
+
     def _require_path(self) -> Path:
         try:
             path = self.path
@@ -933,6 +1006,19 @@ def _require_path(self) -> Path:
             raise RuntimeError("'pattern' must be set")
         return path
 
+    def _check_required_lock(self, path: Path) -> None:
+        if self._locked_path is not None and path != self._locked_path:
+            raise SnapclassError(
+                "Snapshot path changed while locked; keep path fields stable inside "
+                "snapshot.locked()"
+            )
+        if self.require_lock and self._lock_depth == 0:
+            raise SnapclassError(
+                "Snapshot writes require an active snapshot lock; wrap mutation "
+                "and save in `with obj.snapshot.locked():`; use reload=True when "
+                "coordinating shared writers"
+            )
+
     def _check_write_conflict(self, path: Path) -> None:
         if self._config.conflict != "raise" or not path.exists():
             return
@@ -1844,6 +1930,31 @@ def _normalize_conflict_policy(conflict: str) -> str:
     return conflict
 
 
+def _validate_snapshot_pattern(pattern: str | None) -> None:
+    if pattern is None:
+        return
+    _validate_snapshot_path(Path(pattern))
+
+
+def _validate_snapshot_path(path: Path) -> None:
+    if _is_lock_path(path):
+        raise ValueError(
+            "Snapshot filenames cannot end with .lock; .lock is reserved for "
+            "snapclass lock files"
+        )
+
+
+def _validate_require_lock(
+    pattern: str | None,
+    manual: bool,
+    require_lock: bool,
+) -> None:
+    if require_lock and pattern is None:
+        raise ValueError("require_lock=True requires a persisted snapshot pattern")
+    if require_lock and not manual:
+        raise ValueError("require_lock=True requires manual=True")
+
+
 def _validate_extras_field(cls: type, unknown: str, extras_field: str | None) -> None:
     if unknown != "collect":
         return
@@ -2490,7 +2601,11 @@ def iter_candidates(self) -> Iterator[Path]:
         search_root = self.root.joinpath(*static_parts) if static_parts else self.root
         if not search_root.exists():
             return iter(())
-        matches = [path for path in search_root.rglob("*") if self.regex.match(_as_posix(path))]
+        matches = [
+            path
+            for path in search_root.rglob("*")
+            if self.regex.match(_as_posix(path)) and not _is_lock_path(path)
+        ]
         return iter(sorted(matches, key=_as_posix))
 
     def values_from(self, path: Path) -> list[str]:
@@ -2684,17 +2799,8 @@ def _write_text_atomic(path: Path, text: str, *, write_delay: float | None = Non
                 raise
 
 
-def _write_lock_for(path: Path) -> threading.RLock:
-    try:
-        key = path.resolve()
-    except FileNotFoundError:
-        key = path.absolute()
-    with _WRITE_LOCKS_GUARD:
-        lock = _WRITE_LOCKS.get(key)
-        if lock is None:
-            lock = threading.RLock()
-            _WRITE_LOCKS[key] = lock
-        return lock
+def _write_lock_for(path: Path) -> Any:
+    return _shared_write_lock_for(path)
 
 
 def _replace_path_atomic(temp_path: Path, path: Path) -> None:
diff --git a/src/snapclass/sidecar.py b/src/snapclass/sidecar.py
index b1bc217..90ba867 100644
--- a/src/snapclass/sidecar.py
+++ b/src/snapclass/sidecar.py
@@ -229,8 +229,51 @@ def write(
         save_metadata: bool = True,
     ) -> None:
         snapshot = getattr(self._instance, "snapshot", None)
-        if save_metadata and self._field and snapshot is not None:
-            snapshot._check_write_conflict(snapshot._require_path())
+        metadata_path: Path | None = None
+        parent_lock = None
+        needs_lock_check = False
+        if snapshot is not None:
+            needs_lock_check = getattr(snapshot, "require_lock", False) or getattr(
+                snapshot,
+                "_lock_depth",
+                0,
+            )
+            if needs_lock_check or (save_metadata and self._field):
+                metadata_path = snapshot._require_path()
+                from .schemas import _write_lock_for
+
+                parent_lock = _write_lock_for(metadata_path)
+        if parent_lock is not None:
+            with parent_lock:
+                self._check_parent_snapshot_before_write(
+                    snapshot,
+                    metadata_path,
+                    needs_lock_check=needs_lock_check,
+                    save_metadata=save_metadata,
+                )
+                self._write_content(value)
+                self._save_pointer_metadata(save_metadata, snapshot)
+            return
+
+        self._write_content(value)
+        self._save_pointer_metadata(save_metadata, snapshot)
+
+    def _check_parent_snapshot_before_write(
+        self,
+        snapshot: Any,
+        metadata_path: Path | None,
+        *,
+        needs_lock_check: bool,
+        save_metadata: bool,
+    ) -> None:
+        if snapshot is None or metadata_path is None:
+            return
+        if needs_lock_check or (save_metadata and self._field):
+            snapshot._check_required_lock(metadata_path)
+        if save_metadata and self._field:
+            snapshot._check_write_conflict(metadata_path)
+
+    def _write_content(self, value: str | builtins.bytes) -> None:
         self.path.parent.mkdir(parents=True, exist_ok=True)
         if self._descriptor.kind == "text":
             if not isinstance(value, str):
@@ -240,6 +283,12 @@ def write(
             if not isinstance(value, (builtins.bytes, bytearray, memoryview)):
                 raise TypeError("Bytes sidecars require bytes-like values")
             self.path.write_bytes(builtins.bytes(value))
+
+    def _save_pointer_metadata(
+        self,
+        save_metadata: bool,
+        snapshot: Any,
+    ) -> None:
         if self._field:
             object.__setattr__(self._instance, self._field, self.relpath.as_posix())
             if save_metadata and snapshot is not None:
diff --git a/src/snapclass/skills/snapclass-fluency/SKILL.md b/src/snapclass/skills/snapclass-fluency/SKILL.md
index d7b7f41..367b7f5 100644
--- a/src/snapclass/skills/snapclass-fluency/SKILL.md
+++ b/src/snapclass/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
diff --git a/tests/test_collection_patterns.py b/tests/test_collection_patterns.py
index 3f85236..36f3aa6 100644
--- a/tests/test_collection_patterns.py
+++ b/tests/test_collection_patterns.py
@@ -60,6 +60,24 @@ class Prompt:
     ]
 
 
+def test_collection_all_ignores_lock_sidecars_for_extensionless_patterns(tmp_path):
+    root = Stash(tmp_path)
+
+    @snapclass("{self.name}", stash=root, manual=True)
+    class WorkflowState:
+        name: str
+        steps: list[str] = field(default_factory=list)
+
+    state = WorkflowState("run", ["created"])
+    with state.snapshot.locked():
+        state.snapshot.save()
+
+    assert (tmp_path / "run.lock").exists()
+    assert [(item.name, item.steps) for item in WorkflowState.snapshots.all()] == [
+        ("run", ["created"]),
+    ]
+
+
 def test_collection_all_honors_repeated_placeholder_segments(tmp_path):
     root = Stash(tmp_path)
 
diff --git a/tests/test_model_meta.py b/tests/test_model_meta.py
index 4ceb50b..af2fa29 100644
--- a/tests/test_model_meta.py
+++ b/tests/test_model_meta.py
@@ -18,6 +18,7 @@ def test_model_exposes_default_meta_configuration():
     assert Model.Meta.snapshot_write_delay is None
     assert Model.Meta.snapshot_unknown == "ignore"
     assert Model.Meta.snapshot_conflict == "overwrite"
+    assert Model.Meta.snapshot_require_lock is False
 
 
 def test_model_meta_declaration_uses_snapclass_configuration(tmp_path):
@@ -142,6 +143,49 @@ class Meta:
         prompt.snapshot.save()
 
 
+def test_model_meta_snapshot_require_lock_guards_save(tmp_path):
+    class Prompt(Model):
+        name: str
+        body: str = ""
+
+        class Meta:
+            snapshot_pattern = "{self.name}.yml"
+            snapshot_stash = Stash(tmp_path / "prompts")
+            snapshot_manual = True
+            snapshot_require_lock = True
+
+    prompt = Prompt("popsicle", "hello")
+
+    with pytest.raises(SnapclassError, match="active snapshot lock"):
+        prompt.snapshot.save()
+
+    with prompt.snapshot.locked():
+        prompt.snapshot.save()
+
+    assert (tmp_path / "prompts" / "popsicle.yml").read_text(encoding="utf-8") == (
+        "body: hello\n"
+    )
+
+
+def test_model_meta_snapshot_require_lock_requires_pattern():
+    with pytest.raises(ValueError, match="persisted snapshot pattern"):
+        class Prompt(Model):
+            name: str
+
+            class Meta:
+                snapshot_require_lock = True
+
+
+def test_model_meta_rejects_lock_extension_snapshot_pattern():
+    with pytest.raises(ValueError, match="reserved"):
+        class Prompt(Model):
+            name: str
+
+            class Meta:
+                snapshot_pattern = "{self.name}.lock"
+                snapshot_manual = True
+
+
 def test_patternless_model_infers_fields_and_exposes_projection_without_path():
     @dataclass
     class Sample(Model):
@@ -334,6 +378,47 @@ class Prompt:
         prompt.snapshot.save()
 
 
+def test_create_model_accepts_require_lock(tmp_path):
+    @dataclass
+    class Prompt:
+        name: str
+        body: str = ""
+
+    create_model(
+        Prompt,
+        pattern=str(tmp_path / "{self.name}.yml"),
+        manual=True,
+        require_lock=True,
+    )
+    prompt = Prompt("popsicle", "hello")
+
+    with pytest.raises(SnapclassError, match="active snapshot lock"):
+        prompt.snapshot.save()
+
+    with prompt.snapshot.locked():
+        prompt.snapshot.save()
+
+    assert (tmp_path / "popsicle.yml").read_text(encoding="utf-8") == "body: hello\n"
+
+
+def test_create_model_rejects_require_lock_without_pattern():
+    @dataclass
+    class Prompt:
+        name: str
+
+    with pytest.raises(ValueError, match="persisted snapshot pattern"):
+        create_model(Prompt, require_lock=True)
+
+
+def test_create_model_rejects_lock_extension_snapshot_pattern():
+    @dataclass
+    class Prompt:
+        name: str
+
+    with pytest.raises(ValueError, match="reserved"):
+        create_model(Prompt, pattern="{self.name}.lock", manual=True)
+
+
 def test_create_model_rejects_non_dataclass():
     class Plain:
         name: str
diff --git a/tests/test_sidecar.py b/tests/test_sidecar.py
index bfdced0..b250bc6 100644
--- a/tests/test_sidecar.py
+++ b/tests/test_sidecar.py
@@ -3,6 +3,7 @@
 import os
 from dataclasses import fields
 from pathlib import Path
+import threading
 
 import pytest
 
@@ -247,6 +248,122 @@ class Article:
     assert "Human edit" in metadata.read_text(encoding="utf-8")
 
 
+def test_require_lock_blocks_pointer_sidecar_write_before_content_write(tmp_path):
+    articles = Stash(tmp_path / "world") / "article"
+
+    @snapclass(
+        "{self.slug}/article.yml",
+        stash=articles,
+        manual=True,
+        require_lock=True,
+    )
+    class Article:
+        slug: str
+        content_file: str = ""
+        body: str = sidecar.text(field="content_file", default="{self.slug}.md")
+
+    article = Article("dusk-court")
+    metadata = tmp_path / "world" / "article" / "dusk-court" / "article.yml"
+    body = tmp_path / "world" / "article" / "dusk-court" / "dusk-court.md"
+
+    with pytest.raises(SnapclassError, match="active snapshot lock"):
+        article.body = "# Outside lock\n"
+
+    assert article.content_file == ""
+    assert not body.exists()
+
+    with article.snapshot.locked():
+        article.body = "# Inside lock\n"
+
+    assert article.content_file == "dusk-court.md"
+    assert "content_file: dusk-court.md" in metadata.read_text(encoding="utf-8")
+    assert body.read_text(encoding="utf-8") == "# Inside lock\n"
+
+
+def test_require_lock_blocks_competing_thread_sidecar_write_until_lock_released(
+    tmp_path,
+):
+    articles = Stash(tmp_path / "world") / "article"
+
+    @snapclass(
+        "{self.slug}/article.yml",
+        stash=articles,
+        manual=True,
+        require_lock=True,
+    )
+    class Article:
+        slug: str
+        content_file: str = ""
+        body: str = sidecar.text(field="content_file", default="{self.slug}.md")
+
+    article = Article("dusk-court")
+    body = tmp_path / "world" / "article" / "dusk-court" / "dusk-court.md"
+    locked = threading.Event()
+    release = threading.Event()
+    writer_started = threading.Event()
+    writer_done = threading.Event()
+    errors: list[Exception] = []
+
+    def hold_parent_lock() -> None:
+        with article.snapshot.locked():
+            locked.set()
+            assert release.wait(timeout=5)
+
+    def write_sidecar_from_other_thread() -> None:
+        assert locked.wait(timeout=5)
+        writer_started.set()
+        try:
+            article.body = "# Other thread\n"
+        except Exception as exc:
+            errors.append(exc)
+        finally:
+            writer_done.set()
+
+    holder = threading.Thread(target=hold_parent_lock)
+    writer = threading.Thread(target=write_sidecar_from_other_thread)
+    holder.start()
+    writer.start()
+    try:
+        assert locked.wait(timeout=5)
+        assert writer_started.wait(timeout=5)
+        assert not writer_done.wait(timeout=0.2)
+        assert not body.exists()
+    finally:
+        release.set()
+        holder.join(timeout=5)
+        writer.join(timeout=5)
+
+    assert not holder.is_alive()
+    assert not writer.is_alive()
+    assert len(errors) == 1
+    assert isinstance(errors[0], SnapclassError)
+    assert "active snapshot lock" in str(errors[0])
+    assert article.content_file == ""
+    assert not body.exists()
+
+
+def test_require_lock_blocks_constructor_sidecar_write_before_content_write(tmp_path):
+    articles = Stash(tmp_path / "world") / "article"
+
+    @snapclass(
+        "{self.slug}/article.yml",
+        stash=articles,
+        manual=True,
+        require_lock=True,
+    )
+    class Article:
+        slug: str
+        content_file: str = ""
+        body: str = sidecar.text(field="content_file", default="{self.slug}.md")
+
+    body = tmp_path / "world" / "article" / "dusk-court" / "dusk-court.md"
+
+    with pytest.raises(SnapclassError, match="active snapshot lock"):
+        Article("dusk-court", body="# Constructor write\n")
+
+    assert not body.exists()
+
+
 def test_text_sidecar_pointer_stays_relative_after_metadata_move(tmp_path):
     articles = Stash(tmp_path / "world") / "article"
 
diff --git a/tests/test_snapshot_locking.py b/tests/test_snapshot_locking.py
new file mode 100644
index 0000000..58f3123
--- /dev/null
+++ b/tests/test_snapshot_locking.py
@@ -0,0 +1,402 @@
+from __future__ import annotations
+
+from dataclasses import field
+import errno
+import os
+from pathlib import Path
+import subprocess
+import sys
+import time
+
+import pytest
+
+from snapclass import SnapclassError, Stash, snapclass
+from snapclass._locks import (
+    _is_windows_lock_contention,
+    _lock_path_for,
+    _normalized_path,
+    locked_path,
+)
+from snapclass.formatters import YAMLFormatter
+
+
+def _subprocess_env() -> dict[str, str]:
+    env = os.environ.copy()
+    src = Path(__file__).resolve().parents[1] / "src"
+    env["PYTHONPATH"] = os.fspath(src) + os.pathsep + env.get("PYTHONPATH", "")
+    return env
+
+
+def _symlink_or_skip(link: Path, target: Path, *, target_is_directory: bool = False) -> None:
+    try:
+        link.symlink_to(target, target_is_directory=target_is_directory)
+    except (OSError, NotImplementedError) as exc:
+        pytest.skip(f"symlink creation is unavailable: {exc}")
+
+
+def test_lock_normalizes_missing_paths_through_symlinked_parent(tmp_path):
+    real = tmp_path / "real"
+    real.mkdir()
+    link = tmp_path / "link"
+    _symlink_or_skip(link, real, target_is_directory=True)
+
+    normalized = _normalized_path(link / "missing.yml")
+
+    assert normalized == (real / "missing.yml").resolve(strict=False)
+    assert _lock_path_for(normalized) == real / "missing.yml.lock"
+
+
+def test_locked_path_uses_normalized_parent_for_lock_sidecar(tmp_path):
+    real = tmp_path / "real"
+    real.mkdir()
+    link = tmp_path / "link"
+    _symlink_or_skip(link, real, target_is_directory=True)
+
+    with locked_path(link / "state.yml"):
+        assert (real / "state.yml.lock").exists()
+
+    assert (link / "state.yml.lock").resolve() == (real / "state.yml.lock").resolve()
+
+
+def test_locked_path_keeps_leaf_symlink_aligned_with_atomic_replace(tmp_path):
+    real = tmp_path / "real"
+    real.mkdir()
+    target = real / "state.yml"
+    target.write_text("steps:\n  - target\n", encoding="utf-8")
+    alias = tmp_path / "alias.yml"
+    _symlink_or_skip(alias, target)
+
+    normalized = _normalized_path(alias)
+
+    assert normalized == tmp_path / "alias.yml"
+    assert _lock_path_for(normalized) == tmp_path / "alias.yml.lock"
+    with locked_path(alias):
+        assert (tmp_path / "alias.yml.lock").exists()
+
+    assert not (real / "state.yml.lock").exists()
+
+
+def test_snapshot_save_keeps_leaf_symlink_lock_aligned_with_replaced_path(tmp_path):
+    real = tmp_path / "real"
+    real.mkdir()
+    target = real / "state.yml"
+    target.write_text("steps:\n  - target\n", encoding="utf-8")
+    alias = tmp_path / "alias.yml"
+    _symlink_or_skip(alias, target)
+
+    @snapclass("alias.yml", stash=Stash(tmp_path), manual=True)
+    class State:
+        steps: list[str] = field(default_factory=list)
+
+    state = State.snapshots.get()
+    state.steps.append("alias")
+
+    with state.snapshot.locked():
+        state.snapshot.save()
+        assert (tmp_path / "alias.yml.lock").exists()
+        assert _lock_path_for(_normalized_path(alias)) == tmp_path / "alias.yml.lock"
+        assert not (real / "state.yml.lock").exists()
+
+    assert not alias.is_symlink()
+    assert YAMLFormatter.loads(alias.read_text(encoding="utf-8")) == {
+        "steps": ["target", "alias"],
+    }
+    assert YAMLFormatter.loads(target.read_text(encoding="utf-8")) == {
+        "steps": ["target"],
+    }
+
+
+def test_windows_lock_retry_classifier_only_accepts_lock_contention():
+    msvcrt_lock_contention = OSError(errno.EACCES, "permission denied")
+    lock_violation = OSError(errno.EACCES, "locked")
+    lock_violation.winerror = 33
+    sharing_violation = OSError(errno.EACCES, "sharing")
+    sharing_violation.winerror = 32
+    access_denied = OSError(errno.EACCES, "access denied")
+    access_denied.winerror = 5
+    bad_file_descriptor = OSError(errno.EBADF, "bad file descriptor")
+
+    assert _is_windows_lock_contention(msvcrt_lock_contention) is True
+    assert _is_windows_lock_contention(lock_violation) is True
+    assert _is_windows_lock_contention(sharing_violation) is True
+    assert _is_windows_lock_contention(access_denied) is False
+    assert _is_windows_lock_contention(bad_file_descriptor) is False
+
+
+def test_snapshot_locked_reloads_before_mutation_and_saves_inside_block(tmp_path):
+    @snapclass("{self.name}.yml", stash=Stash(tmp_path), manual=True)
+    class WorkflowState:
+        name: str
+        steps: list[str] = field(default_factory=list)
+
+    WorkflowState("run", ["existing"]).snapshot.save()
+    stale = WorkflowState("run", ["stale-local"])
+
+    with stale.snapshot.locked(reload=True):
+        assert stale.steps == ["existing"]
+        stale.steps.append("started")
+        stale.snapshot.save()
+
+    data = YAMLFormatter.loads((tmp_path / "run.yml").read_text(encoding="utf-8"))
+    assert data["steps"] == ["existing", "started"]
+
+
+def test_snapshot_locked_allows_first_save_when_file_is_missing(tmp_path):
+    @snapclass("{self.name}.yml", stash=Stash(tmp_path), manual=True)
+    class WorkflowState:
+        name: str
+        steps: list[str] = field(default_factory=list)
+
+    state = WorkflowState("run")
+
+    with state.snapshot.locked(reload=True):
+        state.steps.append("created")
+        state.snapshot.save()
+
+    assert YAMLFormatter.loads((tmp_path / "run.yml").read_text(encoding="utf-8")) == {
+        "steps": ["created"],
+    }
+
+
+def test_require_lock_blocks_save_outside_locked_context(tmp_path):
+    @snapclass(
+        "{self.name}.yml",
+        stash=Stash(tmp_path),
+        manual=True,
+        require_lock=True,
+    )
+    class WorkflowState:
+        name: str
+        steps: list[str] = field(default_factory=list)
+
+    state = WorkflowState("run")
+    state.steps.append("started")
+
+    with pytest.raises(SnapclassError, match="active snapshot lock"):
+        state.snapshot.save()
+
+    with state.snapshot.locked(reload=True):
+        state.snapshot.save()
+
+    assert YAMLFormatter.loads((tmp_path / "run.yml").read_text(encoding="utf-8")) == {
+        "steps": ["started"],
+    }
+
+
+def test_require_lock_rejects_automatic_models(tmp_path):
+    with pytest.raises(ValueError, match="manual=True"):
+
+        @snapclass("{self.name}.yml", stash=Stash(tmp_path), require_lock=True)
+        class WorkflowState:
+            name: str
+
+
+def test_require_lock_rejects_patternless_snapclass():
+    with pytest.raises(ValueError, match="persisted snapshot pattern"):
+        @snapclass(require_lock=True)
+        class WorkflowState:
+            name: str
+
+
+def test_snapclass_rejects_lock_extension_snapshot_pattern(tmp_path):
+    with pytest.raises(ValueError, match="reserved"):
+
+        @snapclass("{self.name}.lock", stash=Stash(tmp_path), manual=True)
+        class WorkflowState:
+            name: str
+
+
+def test_snapshot_path_rejects_dynamic_lock_extension_filename(tmp_path):
+    @snapclass("{self.name}", stash=Stash(tmp_path), manual=True)
+    class WorkflowState:
+        name: str
+
+    state = WorkflowState("run.lock")
+
+    with pytest.raises(ValueError, match="reserved"):
+        state.snapshot.save()
+
+    with pytest.raises(ValueError, match="reserved"):
+        state.snapshot.path = tmp_path / "manual.lock"
+
+
+def test_snapshot_locked_rejects_path_changes_inside_lock(tmp_path):
+    @snapclass("{self.name}.yml", stash=Stash(tmp_path), manual=True)
+    class WorkflowState:
+        name: str
+        steps: list[str] = field(default_factory=list)
+
+    state = WorkflowState("first")
+
+    with pytest.raises(SnapclassError, match="path changed"):
+        with state.snapshot.locked():
+            state.name = "second"
+            state.snapshot.save()
+
+
+def test_cross_process_locked_reload_preserves_both_updates(tmp_path):
+    script = r"""
+from dataclasses import field
+from pathlib import Path
+import sys
+import time
+
+from snapclass import Stash, snapclass
+
+root = Path(sys.argv[1])
+label = sys.argv[2]
+ready = Path(sys.argv[3])
+start = Path(sys.argv[4])
+
+@snapclass("{self.name}.yml", stash=Stash(root), manual=True, require_lock=True)
+class WorkflowState:
+    name: str
+    steps: list[str] = field(default_factory=list)
+
+ready.write_text("ready", encoding="utf-8")
+while not start.exists():
+    time.sleep(0.01)
+
+state = WorkflowState("shared")
+with state.snapshot.locked(reload=True):
+    state.steps.append(label)
+    time.sleep(0.1)
+    state.snapshot.save()
+"""
+
+    env = _subprocess_env()
+    start = tmp_path / "start"
+    ready_files = [tmp_path / "first.ready", tmp_path / "second.ready"]
+    processes = [
+        subprocess.Popen(
+            [
+                sys.executable,
+                "-c",
+                script,
+                os.fspath(tmp_path),
+                label,
+                os.fspath(ready),
+                os.fspath(start),
+            ],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            env=env,
+        )
+        for label, ready in zip(("lur-id", "article-id"), ready_files)
+    ]
+    try:
+        deadline = time.monotonic() + 10
+        while not all(path.exists() for path in ready_files):
+            assert time.monotonic() < deadline
+            time.sleep(0.01)
+
+        start.write_text("go", encoding="utf-8")
+
+        results = [process.communicate(timeout=15) for process in processes]
+        for process, (stdout, stderr) in zip(processes, results):
+            assert process.returncode == 0, stdout + stderr
+    finally:
+        for process in processes:
+            if process.poll() is None:
+                process.kill()
+
+    data = YAMLFormatter.loads((tmp_path / "shared.yml").read_text(encoding="utf-8"))
+    assert len(data["steps"]) == 2
+    assert set(data["steps"]) == {"lur-id", "article-id"}
+    assert (tmp_path / "shared.yml.lock").exists()
+
+
+def test_process_exit_releases_snapshot_lock(tmp_path):
+    holder_script = r"""
+from dataclasses import field
+from pathlib import Path
+import sys
+import time
+
+from snapclass import Stash, snapclass
+
+root = Path(sys.argv[1])
+ready = Path(sys.argv[2])
+
+@snapclass("{self.name}.yml", stash=Stash(root), manual=True, require_lock=True)
+class WorkflowState:
+    name: str
+    steps: list[str] = field(default_factory=list)
+
+state = WorkflowState("shared")
+with state.snapshot.locked(reload=True):
+    ready.write_text("locked", encoding="utf-8")
+    time.sleep(60)
+"""
+    writer_script = r"""
+from dataclasses import field
+from pathlib import Path
+import sys
+
+from snapclass import Stash, snapclass
+
+root = Path(sys.argv[1])
+
+@snapclass("{self.name}.yml", stash=Stash(root), manual=True, require_lock=True)
+class WorkflowState:
+    name: str
+    steps: list[str] = field(default_factory=list)
+
+state = WorkflowState("shared")
+with state.snapshot.locked(reload=True):
+    state.steps.append("after-kill")
+    state.snapshot.save()
+"""
+    env = _subprocess_env()
+    ready = tmp_path / "holder.ready"
+    holder = subprocess.Popen(
+        [sys.executable, "-c", holder_script, os.fspath(tmp_path), os.fspath(ready)],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+        env=env,
+    )
+    try:
+        deadline = time.monotonic() + 10
+        while not ready.exists():
+            assert time.monotonic() < deadline
+            time.sleep(0.01)
+        holder.kill()
+        holder.communicate(timeout=10)
+
+        writer = subprocess.Popen(
+            [sys.executable, "-c", writer_script, os.fspath(tmp_path)],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            env=env,
+        )
+        stdout, stderr = writer.communicate(timeout=10)
+        assert writer.returncode == 0, stdout + stderr
+    finally:
+        if holder.poll() is None:
+            holder.kill()
+
+    data = YAMLFormatter.loads((tmp_path / "shared.yml").read_text(encoding="utf-8"))
+    assert data["steps"] == ["after-kill"]
+
+
+def test_get_or_create_uses_lock_for_require_lock_models(tmp_path):
+    @snapclass(
+        "{self.name}.yml",
+        stash=Stash(tmp_path),
+        manual=True,
+        defaults=True,
+        require_lock=True,
+    )
+    class WorkflowState:
+        name: str
+        steps: list[str] = field(default_factory=list)
+
+    state = WorkflowState.snapshots.get_or_create("shared")
+
+    assert state.steps == []
+    assert YAMLFormatter.loads((tmp_path / "shared.yml").read_text(encoding="utf-8")) == {
+        "steps": [None],
+    }
diff --git a/tests/test_sync.py b/tests/test_sync.py
index a9cd464..e92e2c0 100644
--- a/tests/test_sync.py
+++ b/tests/test_sync.py
@@ -200,6 +200,40 @@ class Workflow:
         workflow.snapshot.save()
 
 
+def test_sync_accepts_require_lock_for_workflow_snapshots(tmp_path):
+    @dataclass
+    class Workflow:
+        id: str
+        status: str = "created"
+
+    workflow = Workflow("wf-lock", "running")
+    sync(
+        workflow,
+        str(tmp_path / "{self.id}.yml"),
+        manual=True,
+        require_lock=True,
+    )
+
+    with pytest.raises(SnapclassError, match="active snapshot lock"):
+        workflow.snapshot.save()
+
+    with workflow.snapshot.locked():
+        workflow.snapshot.save()
+
+    assert (tmp_path / "wf-lock.yml").read_text(encoding="utf-8") == (
+        "status: running\n"
+    )
+
+
+def test_sync_rejects_lock_extension_snapshot_pattern():
+    @dataclass
+    class Workflow:
+        id: str
+
+    with pytest.raises(ValueError, match="reserved"):
+        sync(Workflow("wf-lock"), "{self.id}.lock", manual=True)
+
+
 def test_sync_snapshot_saves_are_serialized_across_threads(tmp_path):
     @dataclass
     class Step: