diff --git a/AGENTS.md b/AGENTS.md
index d4ed6f3f7..c0be4b3dc 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -180,7 +180,7 @@ Each domain has exactly **one** write-owning module (or one tightly-scoped famil
 
 | Domain | Write-owning module(s) |
 |--------|------------------------|
-| Entities (`entities/*/entity.json`, `entities/*/*.npz`) | `solstone/think/entities/journal.py` + `solstone/think/entities/consolidation.py` + `solstone/think/entities/saving.py` + `solstone/think/entities/merge.py` + `solstone/think/entities/voiceprints.py` + `solstone/apps/entities/call.py` + `solstone/apps/speakers/owner.py` + `solstone/apps/speakers/routes.py` |
+| Entities (`entities/*/entity.json`, `entities/*/*.npz`) | `solstone/think/entities/journal.py` + `solstone/think/entities/relationships.py` + `solstone/think/entities/consolidation.py` + `solstone/think/entities/saving.py` + `solstone/think/entities/merge.py` + `solstone/think/entities/voiceprints.py` + `solstone/apps/entities/call.py` + `solstone/apps/speakers/owner.py` + `solstone/apps/speakers/routes.py` |
 | Owner voice candidate (`awareness/owner_candidate.npz`) | `solstone/apps/speakers/owner.py` |
 | Entity merge candidates (`entities/review-candidates.jsonl`) | `solstone/think/entities/review_candidates.py` + `solstone/apps/entities/call.py` |
 | Facet review candidates (`facets/review-candidates.jsonl`) | `solstone/think/facet_review_candidates.py` |
diff --git a/scripts/check_journal_io_access.py b/scripts/check_journal_io_access.py
index c22b5fddc..2c9309868 100644
--- a/scripts/check_journal_io_access.py
+++ b/scripts/check_journal_io_access.py
@@ -93,6 +93,7 @@ OWNER_FILES: frozenset[str] = frozenset(
         "solstone/think/entities/journal.py",
         "solstone/think/entities/merge.py",
         "solstone/think/entities/observations.py",
+        "solstone/think/entities/relationships.py",
         "solstone/think/entities/review_candidates.py",
         "solstone/think/entities/saving.py",
         "solstone/think/entities/voiceprints.py",
diff --git a/solstone/apps/curation/routes.py b/solstone/apps/curation/routes.py
index 080064ffd..00990859f 100644
--- a/solstone/apps/curation/routes.py
+++ b/solstone/apps/curation/routes.py
@@ -10,7 +10,7 @@ from typing import Any
 from flask import Blueprint, Response, jsonify, render_template, request
 
 from solstone.apps.curation import copy as curation_copy
-from solstone.convey.reasons import MISSING_REQUIRED_FIELD
+from solstone.convey.reasons import ENTITY_BUSY, MISSING_REQUIRED_FIELD
 from solstone.convey.utils import error_response
 from solstone.think.curation import (
     accept_entity_candidate,
@@ -20,6 +20,7 @@ from solstone.think.curation import (
     load_open_items,
     merge_preview_fields,
 )
+from solstone.think.journal_io import LockTimeout
 
 curation_bp = Blueprint("app:curation", __name__, url_prefix="/app/curation")
 
@@ -117,14 +118,18 @@ def accept_entity() -> Response | tuple[Response, int]:
     except KeyError as exc:
         return _missing_field(str(exc.args[0]))
 
-    return _result_response(
-        accept_entity_candidate(
+    try:
+        result = accept_entity_candidate(
             facet,
             source_slug,
             target_slug,
             commit=True,
         )
-    )
+    except LockTimeout:
+        return error_response(
+            ENTITY_BUSY, detail="entity merge candidates are busy; try again"
+        )
+    return _result_response(result)
 
 
 @curation_bp.post("/api/entity/dismiss")
@@ -137,4 +142,10 @@ def dismiss_entity() -> Response | tuple[Response, int]:
     except KeyError as exc:
         return _missing_field(str(exc.args[0]))
 
-    return _result_response(dismiss_entity_candidate(facet, source_slug, target_slug))
+    try:
+        result = dismiss_entity_candidate(facet, source_slug, target_slug)
+    except LockTimeout:
+        return error_response(
+            ENTITY_BUSY, detail="entity merge candidates are busy; try again"
+        )
+    return _result_response(result)
diff --git a/solstone/apps/entities/call.py b/solstone/apps/entities/call.py
index b3c3b7ec2..a87a4bebe 100644
--- a/solstone/apps/entities/call.py
+++ b/solstone/apps/entities/call.py
@@ -13,6 +13,7 @@ from pathlib import Path
 
 import typer
 
+from solstone.convey.reasons import ENTITY_BUSY
 from solstone.think.curation import (
     accept_entity_candidate,
     dismiss_entity_candidate,
@@ -51,6 +52,7 @@ from solstone.think.entities.saving import (
 )
 from solstone.think.facets import log_call_action
 from solstone.think.indexer.journal import search_entities
+from solstone.think.journal_io import LockTimeout
 from solstone.think.utils import (
     get_journal,
     now_ms,
@@ -236,6 +238,9 @@ def detect_entity(
     except ValueError as exc:
         typer.echo(f"Error: {exc}", err=True)
         raise typer.Exit(1)
+    except LockTimeout:
+        typer.echo(ENTITY_BUSY.message, err=True)
+        raise typer.Exit(1)
 
     log_call_action(
         facet=facet,
@@ -367,6 +372,9 @@ def update_entity(
     except ValueError as exc:
         typer.echo(f"Error: {exc}", err=True)
         raise typer.Exit(1)
+    except LockTimeout:
+        typer.echo(ENTITY_BUSY.message, err=True)
+        raise typer.Exit(1)
 
     log_call_action(
         facet=facet,
@@ -527,7 +535,11 @@ def record_merge_candidate(
         created = False
         return rows
 
-    locked_modify_candidates(mutate)
+    try:
+        locked_modify_candidates(mutate)
+    except LockTimeout:
+        typer.echo(ENTITY_BUSY.message, err=True)
+        raise typer.Exit(1)
 
     if row is None:  # pragma: no cover - defensive assertion
         raise RuntimeError("record-merge-candidate produced no row")
@@ -621,12 +633,16 @@ def accept_merge_candidate(
 ) -> None:
     """Preview or accept one recorded entity merge candidate."""
     facet = resolve_sol_facet(facet)
-    result = accept_entity_candidate(
-        facet,
-        source_slug,
-        target_slug,
-        commit=commit,
-    )
+    try:
+        result = accept_entity_candidate(
+            facet,
+            source_slug,
+            target_slug,
+            commit=commit,
+        )
+    except LockTimeout:
+        typer.echo(ENTITY_BUSY.message, err=True)
+        raise typer.Exit(1)
     status = result.get("status")
     if status == "error":
         _echo_merge_candidate_error(result)
@@ -652,7 +668,11 @@ def dismiss_merge_candidate(
 ) -> None:
     """Dismiss one recorded entity merge candidate."""
     facet = resolve_sol_facet(facet)
-    result = dismiss_entity_candidate(facet, source_slug, target_slug)
+    try:
+        result = dismiss_entity_candidate(facet, source_slug, target_slug)
+    except LockTimeout:
+        typer.echo(ENTITY_BUSY.message, err=True)
+        raise typer.Exit(1)
     status = result.get("status")
     if status == "error":
         _echo_merge_candidate_error(result)
@@ -733,6 +753,9 @@ def observe_entity(
     except ValueError as exc:
         typer.echo(f"Error: {exc}", err=True)
         raise typer.Exit(1)
+    except LockTimeout:
+        typer.echo(ENTITY_BUSY.message, err=True)
+        raise typer.Exit(1)
 
     log_call_action(
         facet=facet,
diff --git a/solstone/apps/entities/talent/entity_observer.py b/solstone/apps/entities/talent/entity_observer.py
index c82a1336c..684fa4ca4 100644
--- a/solstone/apps/entities/talent/entity_observer.py
+++ b/solstone/apps/entities/talent/entity_observer.py
@@ -18,6 +18,7 @@ import logging
 from solstone.think.entities.context import assemble_observer_context
 from solstone.think.entities.loading import load_entities
 from solstone.think.entities.observations import add_observation, load_observations
+from solstone.think.journal_io import LockTimeout
 
 logger = logging.getLogger(__name__)
 
@@ -87,7 +88,13 @@ def post_process(result: str, context: dict) -> str | None:
                     "Skipping duplicate observation for %s: %s", entity_id, content[:60]
                 )
                 continue
-            add_observation(facet, entity_id, content, day)
+            try:
+                add_observation(facet, entity_id, content, day)
+            except LockTimeout:
+                logger.warning(
+                    "observations busy for %s; skipping remaining", entity_id
+                )
+                break
             existing.add(content.lower())
 
     return None
diff --git a/solstone/convey/reasons.py b/solstone/convey/reasons.py
index aa5d36bc8..a3bf8ea19 100644
--- a/solstone/convey/reasons.py
+++ b/solstone/convey/reasons.py
@@ -306,6 +306,11 @@ AWARENESS_BUSY = Reason(
     "I couldn't update what I know right now because it was busy. Try again in a moment.",
     503,
 )
+ENTITY_BUSY = Reason(
+    "entity_busy",
+    "I couldn't update that entity right now because it was busy. Try again in a moment.",
+    503,
+)
 
 # reprocess
 REPROCESS_PAST_ONLY = Reason(
diff --git a/solstone/think/entities/__init__.py b/solstone/think/entities/__init__.py
index 9d8c07df0..8ac86887b 100644
--- a/solstone/think/entities/__init__.py
+++ b/solstone/think/entities/__init__.py
@@ -37,7 +37,6 @@ from solstone.think.entities.core import (
     ENTITY_TYPES,
     MAX_ENTITY_SLUG_LENGTH,
     EntityDict,
-    atomic_write,
     entity_last_active_ts,
     entity_slug,
     get_identity_names,
@@ -129,7 +128,6 @@ __all__ = [
     "ENTITY_TYPES",
     "MAX_ENTITY_SLUG_LENGTH",
     "EntityDict",
-    "atomic_write",
     "entity_last_active_ts",
     "entity_slug",
     "get_identity_names",
diff --git a/solstone/think/entities/journal.py b/solstone/think/entities/journal.py
index 475a43b02..7940e8ada 100644
--- a/solstone/think/entities/journal.py
+++ b/solstone/think/entities/journal.py
@@ -15,7 +15,8 @@ import shutil
 from pathlib import Path
 from typing import Any
 
-from solstone.think.entities.core import EntityDict, atomic_write, get_identity_names
+from solstone.think.entities.core import EntityDict, get_identity_names
+from solstone.think.journal_io import atomic_replace
 from solstone.think.utils import get_journal, now_ms
 
 # Global cache for journal entities: {entity_id: EntityDict}
@@ -94,7 +95,7 @@ def save_journal_entity(entity: EntityDict) -> None:
 
     path = journal_entity_path(entity_id)
     content = json.dumps(entity, ensure_ascii=False, indent=2) + "\n"
-    atomic_write(path, content, prefix=".entity_")
+    atomic_replace(path, content)
 
 
 def scan_journal_entities() -> list[str]:
diff --git a/solstone/think/entities/observations.py b/solstone/think/entities/observations.py
index 96c82b8e9..ed0037782 100644
--- a/solstone/think/entities/observations.py
+++ b/solstone/think/entities/observations.py
@@ -10,16 +10,15 @@ They capture useful information like preferences, expertise, relationships,
 and biographical facts that help with future interactions.
 """
 
-import fcntl
 import json
 import random
 import time
 from pathlib import Path
 from typing import Any, Iterator
 
-from solstone.think.entities.core import atomic_write
 from solstone.think.entities.journal import load_all_journal_entities
 from solstone.think.entities.relationships import entity_memory_path
+from solstone.think.journal_io import atomic_replace, hold_lock
 from solstone.think.utils import get_journal, now_ms
 
 # Global cache for entity observations: {(facet, entity_slug): list[dict]}
@@ -213,7 +212,7 @@ def save_observations(
     content = "".join(
         json.dumps(obs, ensure_ascii=False) + "\n" for obs in observations
     )
-    atomic_write(path, content, prefix=".observations_")
+    atomic_replace(path, content)
 
 
 def add_observation(
@@ -251,33 +250,27 @@ def add_observation(
         raise ValueError("Observation content cannot be empty")
 
     path = observations_file_path(facet, name)
-    lock_path = path.parent / f"{path.name}.lock"
 
     last_error: Exception | None = None
     for attempt in range(max_retries):
         try:
-            path.parent.mkdir(parents=True, exist_ok=True)
-            with open(lock_path, "w") as lock_file:
-                fcntl.flock(lock_file, fcntl.LOCK_EX)
-                try:
-                    observations = load_observations(facet, name)
-
-                    observation: dict[str, Any] = {
-                        "content": content,
-                        "observed_at": now_ms(),
-                    }
-                    if source_day:
-                        observation["source_day"] = source_day
-
-                    observations.append(observation)
-                    save_observations(facet, name, observations)
-
-                    return {
-                        "observations": observations,
-                        "count": len(observations),
-                    }
-                finally:
-                    fcntl.flock(lock_file, fcntl.LOCK_UN)
+            with hold_lock(path):
+                observations = load_observations(facet, name)
+
+                observation: dict[str, Any] = {
+                    "content": content,
+                    "observed_at": now_ms(),
+                }
+                if source_day:
+                    observation["source_day"] = source_day
+
+                observations.append(observation)
+                save_observations(facet, name, observations)
+
+                return {
+                    "observations": observations,
+                    "count": len(observations),
+                }
         except ValueError:
             raise  # Logical errors — don't retry
         except OSError as exc:
diff --git a/solstone/think/entities/relationships.py b/solstone/think/entities/relationships.py
index 904fda5bf..52f49d495 100644
--- a/solstone/think/entities/relationships.py
+++ b/solstone/think/entities/relationships.py
@@ -17,7 +17,8 @@ import json
 import shutil
 from pathlib import Path
 
-from solstone.think.entities.core import EntityDict, atomic_write, entity_slug
+from solstone.think.entities.core import EntityDict, entity_slug
+from solstone.think.journal_io import atomic_replace
 from solstone.think.utils import get_journal
 
 # Global cache for facet relationships: {(facet, entity_id): EntityDict}
@@ -105,7 +106,7 @@ def save_facet_relationship(
     relationship["entity_id"] = entity_id
 
     content = json.dumps(relationship, ensure_ascii=False, indent=2) + "\n"
-    atomic_write(path, content, prefix=".relationship_")
+    atomic_replace(path, content)
 
 
 def scan_facet_relationships(facet: str) -> list[str]:
diff --git a/solstone/think/entities/review_candidates.py b/solstone/think/entities/review_candidates.py
index ee0566433..43fe67794 100644
--- a/solstone/think/entities/review_candidates.py
+++ b/solstone/think/entities/review_candidates.py
@@ -8,14 +8,13 @@ Sole write-owner of:
 
 from __future__ import annotations
 
-import fcntl
 import json
 import logging
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Callable
 
-from solstone.think.entities.core import atomic_write
+from solstone.think.journal_io import atomic_replace, hold_lock
 from solstone.think.utils import get_journal
 
 logger = logging.getLogger(__name__)
@@ -33,11 +32,6 @@ def review_candidates_path() -> Path:
     return review_candidates_dir() / "review-candidates.jsonl"
 
 
-def review_candidates_lock_path() -> Path:
-    """Return the sibling lock path for review-candidates.jsonl."""
-    return review_candidates_dir() / ".review-candidates.lock"
-
-
 def _load_jsonl_rows(path: Path) -> list[dict[str, Any]]:
     """Load JSONL rows from *path*, skipping blanks and malformed lines."""
     if not path.exists():
@@ -78,7 +72,7 @@ def _save_jsonl_rows(path: Path, rows: list[dict[str, Any]]) -> None:
     content = ""
     if rows:
         content = "\n".join(json.dumps(row, ensure_ascii=False) for row in rows) + "\n"
-    atomic_write(path, content)
+    atomic_replace(path, content)
 
 
 def save_candidates(rows: list[dict[str, Any]]) -> None:
@@ -111,18 +105,11 @@ def locked_modify_candidates(
     fn: Callable[[list[dict[str, Any]]], list[dict[str, Any]]],
 ) -> list[dict[str, Any]]:
     """Apply a locked read-modify-write cycle to review-candidates.jsonl."""
-    review_candidates_dir()
-    lock_path = review_candidates_lock_path()
-    # Lock file contents are irrelevant; opening with "w" matches the existing pattern.
-    with open(lock_path, "w", encoding="utf-8") as lock_file:
-        fcntl.flock(lock_file, fcntl.LOCK_EX)
-        try:
-            rows = load_candidates()
-            new_rows = fn(rows)
-            save_candidates(new_rows)
-            return new_rows
-        finally:
-            fcntl.flock(lock_file, fcntl.LOCK_UN)
+    with hold_lock(review_candidates_path()):
+        rows = load_candidates()
+        new_rows = fn(rows)
+        save_candidates(new_rows)
+        return new_rows
 
 
 def utc_now_iso() -> str:
diff --git a/solstone/think/entities/saving.py b/solstone/think/entities/saving.py
index 79e0a74ef..a9973a7de 100644
--- a/solstone/think/entities/saving.py
+++ b/solstone/think/entities/saving.py
@@ -8,12 +8,11 @@ This module handles saving entities to storage:
 - save_detected_entity: Concurrency-safe single entity detection with file locking
 """
 
-import fcntl
 import json
 import random
 import time
 
-from solstone.think.entities.core import EntityDict, atomic_write, entity_slug
+from solstone.think.entities.core import EntityDict, entity_slug
 from solstone.think.entities.journal import (
     create_journal_entity,
     load_journal_entity,
@@ -25,6 +24,7 @@ from solstone.think.entities.loading import (
     load_entities,
 )
 from solstone.think.entities.relationships import save_facet_relationship
+from solstone.think.journal_io import atomic_replace, hold_lock
 
 
 def _save_entities_detected(facet: str, entities: list[EntityDict], day: str) -> None:
@@ -45,7 +45,7 @@ def _save_entities_detected(facet: str, entities: list[EntityDict], day: str) ->
 
     # Format as JSONL and write atomically
     content = "".join(json.dumps(e, ensure_ascii=False) + "\n" for e in sorted_entities)
-    atomic_write(path, content, prefix="entities_")
+    atomic_replace(path, content)
     clear_entity_loading_cache()
 
 
@@ -208,23 +208,17 @@ def _locked_modify_detected(
         OSError: If all retries exhausted on transient errors
     """
     path = detected_entities_path(facet, day)
-    lock_path = path.parent / f"{path.name}.lock"
 
     last_error: Exception | None = None
     for attempt in range(max_retries):
         try:
-            path.parent.mkdir(parents=True, exist_ok=True)
-            with open(lock_path, "w") as lock_file:
-                fcntl.flock(lock_file, fcntl.LOCK_EX)
-                try:
-                    # Fresh load inside lock — sees all prior writers' changes
-                    clear_entity_loading_cache()
-                    entities = load_entities(facet, day)
-                    entities = modify_fn(entities)
-                    _save_entities_detected(facet, entities, day)
-                    return entities
-                finally:
-                    fcntl.flock(lock_file, fcntl.LOCK_UN)
+            with hold_lock(path):
+                # Fresh load inside lock — sees all prior writers' changes
+                clear_entity_loading_cache()
+                entities = load_entities(facet, day)
+                entities = modify_fn(entities)
+                _save_entities_detected(facet, entities, day)
+                return entities
         except ValueError:
             raise  # Logical errors (duplicate, not found) — don't retry
         except OSError as exc:
diff --git a/solstone/think/entities/seeding.py b/solstone/think/entities/seeding.py
index 1412c9378..9650e33bc 100644
--- a/solstone/think/entities/seeding.py
+++ b/solstone/think/entities/seeding.py
@@ -9,11 +9,16 @@ This module handles seeding entities from structured imports:
 
 from __future__ import annotations
 
+import logging
 from typing import TYPE_CHECKING
 
+from solstone.think.journal_io import LockTimeout
+
 if TYPE_CHECKING:
     from solstone.think.entities.core import EntityDict
 
+logger = logging.getLogger(__name__)
+
 
 def seed_entities(
     facet: str,
@@ -109,7 +114,16 @@ def seed_entities(
 
                         ensure_facet(facet)
                         facet_ensured = True
-                    add_observation(facet, resolved_name, obs_content, source_day=day)
+                    try:
+                        add_observation(
+                            facet, resolved_name, obs_content, source_day=day
+                        )
+                    except LockTimeout:
+                        logger.warning(
+                            "observations busy for %s; skipping remaining",
+                            resolved_name,
+                        )
+                        break
                     existing_contents.add(obs_content)
 
     return resolved
diff --git a/tests/test_entities.py b/tests/test_entities.py
index 29e7420d4..c26569ec9 100644
--- a/tests/test_entities.py
+++ b/tests/test_entities.py
@@ -409,27 +409,27 @@ def test_save_detected_entity_retry_on_error(fixture_journal, tmp_path, monkeypa
     """Test that save_detected_entity retries on transient OSError."""
     from unittest.mock import patch
 
+    from solstone.think.journal_io import atomic_replace as _real_atomic_replace
+
     monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
     (tmp_path / "facets" / "test_facet" / "entities").mkdir(parents=True)
 
     call_count = 0
-    original_atomic_write = __import__(
-        "solstone.think.entities.core", fromlist=["atomic_write"]
-    ).atomic_write
 
-    def flaky_atomic_write(path, content, prefix=".tmp_"):
+    def flaky_atomic_replace(path, data, *, mode=None):
         nonlocal call_count
         call_count += 1
         if call_count == 1:
             raise PermissionError("Simulated transient error")
-        return original_atomic_write(path, content, prefix)
+        return _real_atomic_replace(path, data, mode=mode)
 
     with patch(
-        "solstone.think.entities.saving.atomic_write", side_effect=flaky_atomic_write
+        "solstone.think.entities.saving.atomic_replace",
+        side_effect=flaky_atomic_replace,
     ):
         save_detected_entity("test_facet", "20250101", "Person", "Alice", "Friend")
 
-    assert call_count == 2  # First attempt failed, second succeeded
+    assert call_count == 2
     loaded = load_entities("test_facet", "20250101")
     assert len(loaded) == 1
     assert loaded[0]["name"] == "Alice"
@@ -2074,6 +2074,34 @@ def test_add_observation_success(fixture_journal, tmp_path, monkeypatch):
     assert len(loaded) == 2
 
 
+def test_add_observation_retry_on_error(fixture_journal, tmp_path, monkeypatch):
+    """Test that add_observation retries on transient OSError."""
+    from unittest.mock import patch
+
+    from solstone.think.journal_io import atomic_replace as _real_atomic_replace
+
+    monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
+
+    call_count = 0
+
+    def flaky_atomic_replace(path, data, *, mode=None):
+        nonlocal call_count
+        call_count += 1
+        if call_count == 1:
+            raise PermissionError("Simulated transient error")
+        return _real_atomic_replace(path, data, mode=mode)
+
+    with patch(
+        "solstone.think.entities.observations.atomic_replace",
+        side_effect=flaky_atomic_replace,
+    ):
+        add_observation("personal", "Alice", "Prefers async communication", "20250113")
+
+    assert call_count == 2
+    loaded = load_observations("personal", "Alice")
+    assert [obs["content"] for obs in loaded] == ["Prefers async communication"]
+
+
 def test_add_observation_empty_content(fixture_journal, tmp_path, monkeypatch):
     """Test adding observation with empty content fails."""
     monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
diff --git a/tests/test_entities_locking.py b/tests/test_entities_locking.py
new file mode 100644
index 000000000..9d1f1c49f
--- /dev/null
+++ b/tests/test_entities_locking.py
@@ -0,0 +1,164 @@
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+from __future__ import annotations
+
+import multiprocessing
+import os
+import traceback
+from pathlib import Path
+from queue import Empty
+from typing import Any
+
+
+def _observation_worker(
+    journal_path: str,
+    barrier: Any,
+    errors: Any,
+    index: int,
+) -> None:
+    os.environ["SOLSTONE_JOURNAL"] = journal_path
+    try:
+        from solstone.think.entities.observations import add_observation
+
+        barrier.wait(timeout=5)
+        add_observation("work", "Alice", f"obs-{index}", source_day="20250101")
+    except BaseException:
+        errors.put(traceback.format_exc())
+        raise
+
+
+def _detected_entity_worker(
+    journal_path: str,
+    barrier: Any,
+    errors: Any,
+    index: int,
+) -> None:
+    os.environ["SOLSTONE_JOURNAL"] = journal_path
+    try:
+        from solstone.think.entities.saving import save_detected_entity
+
+        barrier.wait(timeout=5)
+        save_detected_entity("work", "20250101", "Person", f"E{index}", "d")
+    except BaseException:
+        errors.put(traceback.format_exc())
+        raise
+
+
+def _candidate_worker(
+    journal_path: str,
+    barrier: Any,
+    errors: Any,
+    index: int,
+) -> None:
+    os.environ["SOLSTONE_JOURNAL"] = journal_path
+    try:
+        from solstone.think.entities.review_candidates import locked_modify_candidates
+
+        def mutate(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+            next_rows = list(rows)
+            next_rows.append(
+                {
+                    "facet": "work",
+                    "source_slug": f"s{index}",
+                    "target_slug": "target",
+                }
+            )
+            return next_rows
+
+        barrier.wait(timeout=5)
+        locked_modify_candidates(mutate)
+    except BaseException:
+        errors.put(traceback.format_exc())
+        raise
+
+
+def _drain_errors(errors: Any) -> list[str]:
+    found = []
+    while True:
+        try:
+            found.append(errors.get_nowait())
+        except Empty:
+            return found
+
+
+def _join_processes(processes: list[Any], errors: Any) -> None:
+    for process in processes:
+        process.start()
+    for process in processes:
+        process.join(timeout=10)
+    for process in processes:
+        if process.is_alive():
+            process.terminate()
+            process.join(timeout=2)
+
+    error_text = "\n".join(_drain_errors(errors))
+    assert all(not process.is_alive() for process in processes), error_text
+    assert all(process.exitcode == 0 for process in processes), error_text
+
+
+def _run_workers(tmp_path: Path, monkeypatch: Any, name: str, target: Any) -> Path:
+    ctx = multiprocessing.get_context("spawn")
+    journal = tmp_path / name
+    monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
+    barrier = ctx.Barrier(4)
+    errors = ctx.Queue()
+    processes = [
+        ctx.Process(target=target, args=(str(journal), barrier, errors, i))
+        for i in range(4)
+    ]
+
+    _join_processes(processes, errors)
+    return journal
+
+
+def test_add_observation_serializes_process_writers(
+    tmp_path: Path,
+    monkeypatch: Any,
+) -> None:
+    _run_workers(tmp_path, monkeypatch, "observations", _observation_worker)
+
+    from solstone.think.entities.observations import (
+        clear_observation_cache,
+        load_observations,
+    )
+
+    clear_observation_cache()
+    observations = load_observations("work", "Alice")
+
+    assert sorted(obs["content"] for obs in observations) == [
+        "obs-0",
+        "obs-1",
+        "obs-2",
+        "obs-3",
+    ]
+
+
+def test_save_detected_entity_serializes_process_writers(
+    tmp_path: Path,
+    monkeypatch: Any,
+) -> None:
+    _run_workers(tmp_path, monkeypatch, "detected", _detected_entity_worker)
+
+    from solstone.think.entities.loading import (
+        clear_entity_loading_cache,
+        load_entities,
+    )
+
+    clear_entity_loading_cache()
+    entities = load_entities("work", "20250101")
+
+    assert sorted(entity["name"] for entity in entities) == ["E0", "E1", "E2", "E3"]
+
+
+def test_locked_modify_candidates_serializes_process_writers(
+    tmp_path: Path,
+    monkeypatch: Any,
+) -> None:
+    _run_workers(tmp_path, monkeypatch, "candidates", _candidate_worker)
+
+    from solstone.think.entities.review_candidates import load_candidates
+
+    rows = load_candidates()
+
+    assert sorted(row["source_slug"] for row in rows) == ["s0", "s1", "s2", "s3"]
diff --git a/tests/test_think_review_candidates.py b/tests/test_think_review_candidates.py
index 37ef9983b..41fa7a1f2 100644
--- a/tests/test_think_review_candidates.py
+++ b/tests/test_think_review_candidates.py
@@ -6,6 +6,7 @@ from __future__ import annotations
 import logging
 import threading
 from pathlib import Path
+from unittest.mock import patch
 
 import pytest
 
@@ -17,7 +18,6 @@ from solstone.think.entities.review_candidates import (
     load_candidates,
     locked_modify_candidates,
     review_candidates_dir,
-    review_candidates_lock_path,
     review_candidates_path,
     save_candidates,
     touch_updated,
@@ -44,10 +44,6 @@ def test_path_helpers_return_expected_names(candidate_journal):
         review_candidates_path()
         == candidate_journal / "entities" / "review-candidates.jsonl"
     )
-    assert (
-        review_candidates_lock_path()
-        == candidate_journal / "entities" / ".review-candidates.lock"
-    )
 
 
 def test_load_candidates_missing_file_returns_empty(candidate_journal):
@@ -203,6 +199,20 @@ def test_locked_modify_candidates_applies_fn_and_persists(candidate_journal):
     ]
 
 
+def test_locked_modify_candidates_does_not_retry_on_write_error(candidate_journal):
+    def mutate(rows):
+        return list(rows) + [{"facet": "work", "source_slug": "s", "target_slug": "t"}]
+
+    with patch(
+        "solstone.think.entities.review_candidates.atomic_replace",
+        side_effect=PermissionError("Simulated write error"),
+    ) as atomic_replace:
+        with pytest.raises(OSError):
+            locked_modify_candidates(mutate)
+
+    assert atomic_replace.call_count == 1
+
+
 def test_locked_modify_candidates_serializes_threads(candidate_journal):
     barrier = threading.Barrier(4)
     exceptions: list[BaseException] = []