solpbc.org / solstone
personal memory agent
0
Fork 0

Configure Feed

Select the types of activity you want to include in your feed.

refactor(entities): single-source entity I/O on journal_io primitives

Migrate the entity/observation domain's write mechanics onto the shared
solstone.think.journal_io primitives.

- Five writers (save_journal_entity, save_facet_relationship,
_save_entities_detected, save_observations, review-candidate
_save_jsonl_rows) move from core.atomic_write to atomic_replace, preserving
byte-exact serialization (ensure_ascii=False, indent, trailing newline) and
gaining fsync durability.
- Three locked RMWs (add_observation, _locked_modify_detected,
locked_modify_candidates) route through hold_lock, each keeping its existing
retry policy (observations + detected: 3x OSError retry; candidates: no
retry). LockTimeout (a RuntimeError) propagates past the OSError loops by
design. The bespoke review-candidates lock sidecar is replaced by hold_lock's
convention; review_candidates_lock_path() is removed.
- Surface the new LockTimeout failure mode as ENTITY_BUSY (503, owner-voice)
across the CLI entities verbs, the curation accept/dismiss entity routes, and
the entity-observer talent + seeding (log-and-skip).
- Register relationships.py as an Entities write-owner (AGENTS.md L2 +
check_journal_io_access.py OWNER_FILES); drop the dead atomic_write re-export
from entities/__init__.py.
- Tests: repoint the mechanic-patching retry test to atomic_replace, add an
observations retry test and a candidate no-retry test, and add a cross-process
serialization test covering all three RMWs.

core.atomic_write is retained for its six out-of-domain callers (deletion
deferred to a follow-up lode). merge.py _apply_segment_plan's non-atomic
temp-write is flagged for R3.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

author
Jer Miller co-author
Claude Opus 4.8 (1M context) date (Jun 6, 2026, 9:10 PM -0600) commit 1a2c5f64 parent 46146e9e
+330 -93
+1 -1
AGENTS.md
··· 180 180 181 181 | Domain | Write-owning module(s) | 182 182 |--------|------------------------| 183 - | 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` | 183 + | 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` | 184 184 | Owner voice candidate (`awareness/owner_candidate.npz`) | `solstone/apps/speakers/owner.py` | 185 185 | Entity merge candidates (`entities/review-candidates.jsonl`) | `solstone/think/entities/review_candidates.py` + `solstone/apps/entities/call.py` | 186 186 | Facet review candidates (`facets/review-candidates.jsonl`) | `solstone/think/facet_review_candidates.py` |
+1
scripts/check_journal_io_access.py
··· 93 93 "solstone/think/entities/journal.py", 94 94 "solstone/think/entities/merge.py", 95 95 "solstone/think/entities/observations.py", 96 + "solstone/think/entities/relationships.py", 96 97 "solstone/think/entities/review_candidates.py", 97 98 "solstone/think/entities/saving.py", 98 99 "solstone/think/entities/voiceprints.py",
+16 -5
solstone/apps/curation/routes.py
··· 10 10 from flask import Blueprint, Response, jsonify, render_template, request 11 11 12 12 from solstone.apps.curation import copy as curation_copy 13 - from solstone.convey.reasons import MISSING_REQUIRED_FIELD 13 + from solstone.convey.reasons import ENTITY_BUSY, MISSING_REQUIRED_FIELD 14 14 from solstone.convey.utils import error_response 15 15 from solstone.think.curation import ( 16 16 accept_entity_candidate, ··· 20 20 load_open_items, 21 21 merge_preview_fields, 22 22 ) 23 + from solstone.think.journal_io import LockTimeout 23 24 24 25 curation_bp = Blueprint("app:curation", __name__, url_prefix="/app/curation") 25 26 ··· 117 118 except KeyError as exc: 118 119 return _missing_field(str(exc.args[0])) 119 120 120 - return _result_response( 121 - accept_entity_candidate( 121 + try: 122 + result = accept_entity_candidate( 122 123 facet, 123 124 source_slug, 124 125 target_slug, 125 126 commit=True, 126 127 ) 127 - ) 128 + except LockTimeout: 129 + return error_response( 130 + ENTITY_BUSY, detail="entity merge candidates are busy; try again" 131 + ) 132 + return _result_response(result) 128 133 129 134 130 135 @curation_bp.post("/api/entity/dismiss") ··· 137 142 except KeyError as exc: 138 143 return _missing_field(str(exc.args[0])) 139 144 140 - return _result_response(dismiss_entity_candidate(facet, source_slug, target_slug)) 145 + try: 146 + result = dismiss_entity_candidate(facet, source_slug, target_slug) 147 + except LockTimeout: 148 + return error_response( 149 + ENTITY_BUSY, detail="entity merge candidates are busy; try again" 150 + ) 151 + return _result_response(result)
+31 -8
solstone/apps/entities/call.py
··· 13 13 14 14 import typer 15 15 16 + from solstone.convey.reasons import ENTITY_BUSY 16 17 from solstone.think.curation import ( 17 18 accept_entity_candidate, 18 19 dismiss_entity_candidate, ··· 51 52 ) 52 53 from solstone.think.facets import log_call_action 53 54 from solstone.think.indexer.journal import search_entities 55 + from solstone.think.journal_io import LockTimeout 54 56 from solstone.think.utils import ( 55 57 get_journal, 56 58 now_ms, ··· 235 237 save_detected_entity(facet, day, type_, name, description) 236 238 except ValueError as exc: 237 239 typer.echo(f"Error: {exc}", err=True) 240 + raise typer.Exit(1) 241 + except LockTimeout: 242 + typer.echo(ENTITY_BUSY.message, err=True) 238 243 raise typer.Exit(1) 239 244 240 245 log_call_action( ··· 367 372 except ValueError as exc: 368 373 typer.echo(f"Error: {exc}", err=True) 369 374 raise typer.Exit(1) 375 + except LockTimeout: 376 + typer.echo(ENTITY_BUSY.message, err=True) 377 + raise typer.Exit(1) 370 378 371 379 log_call_action( 372 380 facet=facet, ··· 527 535 created = False 528 536 return rows 529 537 530 - locked_modify_candidates(mutate) 538 + try: 539 + locked_modify_candidates(mutate) 540 + except LockTimeout: 541 + typer.echo(ENTITY_BUSY.message, err=True) 542 + raise typer.Exit(1) 531 543 532 544 if row is None: # pragma: no cover - defensive assertion 533 545 raise RuntimeError("record-merge-candidate produced no row") ··· 621 633 ) -> None: 622 634 """Preview or accept one recorded entity merge candidate.""" 623 635 facet = resolve_sol_facet(facet) 624 - result = accept_entity_candidate( 625 - facet, 626 - source_slug, 627 - target_slug, 628 - commit=commit, 629 - ) 636 + try: 637 + result = accept_entity_candidate( 638 + facet, 639 + source_slug, 640 + target_slug, 641 + commit=commit, 642 + ) 643 + except LockTimeout: 644 + typer.echo(ENTITY_BUSY.message, err=True) 645 + raise typer.Exit(1) 630 646 status = result.get("status") 631 647 if status == "error": 632 648 _echo_merge_candidate_error(result) ··· 652 668 ) -> None: 653 669 """Dismiss one recorded entity merge candidate.""" 654 670 facet = resolve_sol_facet(facet) 655 - result = dismiss_entity_candidate(facet, source_slug, target_slug) 671 + try: 672 + result = dismiss_entity_candidate(facet, source_slug, target_slug) 673 + except LockTimeout: 674 + typer.echo(ENTITY_BUSY.message, err=True) 675 + raise typer.Exit(1) 656 676 status = result.get("status") 657 677 if status == "error": 658 678 _echo_merge_candidate_error(result) ··· 732 752 add_observation(facet, resolved_name, content, source_day) 733 753 except ValueError as exc: 734 754 typer.echo(f"Error: {exc}", err=True) 755 + raise typer.Exit(1) 756 + except LockTimeout: 757 + typer.echo(ENTITY_BUSY.message, err=True) 735 758 raise typer.Exit(1) 736 759 737 760 log_call_action(
+8 -1
solstone/apps/entities/talent/entity_observer.py
··· 18 18 from solstone.think.entities.context import assemble_observer_context 19 19 from solstone.think.entities.loading import load_entities 20 20 from solstone.think.entities.observations import add_observation, load_observations 21 + from solstone.think.journal_io import LockTimeout 21 22 22 23 logger = logging.getLogger(__name__) 23 24 ··· 87 88 "Skipping duplicate observation for %s: %s", entity_id, content[:60] 88 89 ) 89 90 continue 90 - add_observation(facet, entity_id, content, day) 91 + try: 92 + add_observation(facet, entity_id, content, day) 93 + except LockTimeout: 94 + logger.warning( 95 + "observations busy for %s; skipping remaining", entity_id 96 + ) 97 + break 91 98 existing.add(content.lower()) 92 99 93 100 return None
+5
solstone/convey/reasons.py
··· 306 306 "I couldn't update what I know right now because it was busy. Try again in a moment.", 307 307 503, 308 308 ) 309 + ENTITY_BUSY = Reason( 310 + "entity_busy", 311 + "I couldn't update that entity right now because it was busy. Try again in a moment.", 312 + 503, 313 + ) 309 314 310 315 # reprocess 311 316 REPROCESS_PAST_ONLY = Reason(
-2
solstone/think/entities/__init__.py
··· 37 37 ENTITY_TYPES, 38 38 MAX_ENTITY_SLUG_LENGTH, 39 39 EntityDict, 40 - atomic_write, 41 40 entity_last_active_ts, 42 41 entity_slug, 43 42 get_identity_names, ··· 129 128 "ENTITY_TYPES", 130 129 "MAX_ENTITY_SLUG_LENGTH", 131 130 "EntityDict", 132 - "atomic_write", 133 131 "entity_last_active_ts", 134 132 "entity_slug", 135 133 "get_identity_names",
+3 -2
solstone/think/entities/journal.py
··· 15 15 from pathlib import Path 16 16 from typing import Any 17 17 18 - from solstone.think.entities.core import EntityDict, atomic_write, get_identity_names 18 + from solstone.think.entities.core import EntityDict, get_identity_names 19 + from solstone.think.journal_io import atomic_replace 19 20 from solstone.think.utils import get_journal, now_ms 20 21 21 22 # Global cache for journal entities: {entity_id: EntityDict} ··· 94 95 95 96 path = journal_entity_path(entity_id) 96 97 content = json.dumps(entity, ensure_ascii=False, indent=2) + "\n" 97 - atomic_write(path, content, prefix=".entity_") 98 + atomic_replace(path, content) 98 99 99 100 100 101 def scan_journal_entities() -> list[str]:
+16 -23
solstone/think/entities/observations.py
··· 10 10 and biographical facts that help with future interactions. 11 11 """ 12 12 13 - import fcntl 14 13 import json 15 14 import random 16 15 import time 17 16 from pathlib import Path 18 17 from typing import Any, Iterator 19 18 20 - from solstone.think.entities.core import atomic_write 21 19 from solstone.think.entities.journal import load_all_journal_entities 22 20 from solstone.think.entities.relationships import entity_memory_path 21 + from solstone.think.journal_io import atomic_replace, hold_lock 23 22 from solstone.think.utils import get_journal, now_ms 24 23 25 24 # Global cache for entity observations: {(facet, entity_slug): list[dict]} ··· 213 212 content = "".join( 214 213 json.dumps(obs, ensure_ascii=False) + "\n" for obs in observations 215 214 ) 216 - atomic_write(path, content, prefix=".observations_") 215 + atomic_replace(path, content) 217 216 218 217 219 218 def add_observation( ··· 251 250 raise ValueError("Observation content cannot be empty") 252 251 253 252 path = observations_file_path(facet, name) 254 - lock_path = path.parent / f"{path.name}.lock" 255 253 256 254 last_error: Exception | None = None 257 255 for attempt in range(max_retries): 258 256 try: 259 - path.parent.mkdir(parents=True, exist_ok=True) 260 - with open(lock_path, "w") as lock_file: 261 - fcntl.flock(lock_file, fcntl.LOCK_EX) 262 - try: 263 - observations = load_observations(facet, name) 257 + with hold_lock(path): 258 + observations = load_observations(facet, name) 264 259 265 - observation: dict[str, Any] = { 266 - "content": content, 267 - "observed_at": now_ms(), 268 - } 269 - if source_day: 270 - observation["source_day"] = source_day 260 + observation: dict[str, Any] = { 261 + "content": content, 262 + "observed_at": now_ms(), 263 + } 264 + if source_day: 265 + observation["source_day"] = source_day 271 266 272 - observations.append(observation) 273 - save_observations(facet, name, observations) 267 + observations.append(observation) 268 + save_observations(facet, name, observations) 274 269 275 - return { 276 - "observations": observations, 277 - "count": len(observations), 278 - } 279 - finally: 280 - fcntl.flock(lock_file, fcntl.LOCK_UN) 270 + return { 271 + "observations": observations, 272 + "count": len(observations), 273 + } 281 274 except ValueError: 282 275 raise # Logical errors — don't retry 283 276 except OSError as exc:
+3 -2
solstone/think/entities/relationships.py
··· 17 17 import shutil 18 18 from pathlib import Path 19 19 20 - from solstone.think.entities.core import EntityDict, atomic_write, entity_slug 20 + from solstone.think.entities.core import EntityDict, entity_slug 21 + from solstone.think.journal_io import atomic_replace 21 22 from solstone.think.utils import get_journal 22 23 23 24 # Global cache for facet relationships: {(facet, entity_id): EntityDict} ··· 105 106 relationship["entity_id"] = entity_id 106 107 107 108 content = json.dumps(relationship, ensure_ascii=False, indent=2) + "\n" 108 - atomic_write(path, content, prefix=".relationship_") 109 + atomic_replace(path, content) 109 110 110 111 111 112 def scan_facet_relationships(facet: str) -> list[str]:
+7 -20
solstone/think/entities/review_candidates.py
··· 8 8 9 9 from __future__ import annotations 10 10 11 - import fcntl 12 11 import json 13 12 import logging 14 13 from datetime import datetime, timezone 15 14 from pathlib import Path 16 15 from typing import Any, Callable 17 16 18 - from solstone.think.entities.core import atomic_write 17 + from solstone.think.journal_io import atomic_replace, hold_lock 19 18 from solstone.think.utils import get_journal 20 19 21 20 logger = logging.getLogger(__name__) ··· 33 32 return review_candidates_dir() / "review-candidates.jsonl" 34 33 35 34 36 - def review_candidates_lock_path() -> Path: 37 - """Return the sibling lock path for review-candidates.jsonl.""" 38 - return review_candidates_dir() / ".review-candidates.lock" 39 - 40 - 41 35 def _load_jsonl_rows(path: Path) -> list[dict[str, Any]]: 42 36 """Load JSONL rows from *path*, skipping blanks and malformed lines.""" 43 37 if not path.exists(): ··· 78 72 content = "" 79 73 if rows: 80 74 content = "\n".join(json.dumps(row, ensure_ascii=False) for row in rows) + "\n" 81 - atomic_write(path, content) 75 + atomic_replace(path, content) 82 76 83 77 84 78 def save_candidates(rows: list[dict[str, Any]]) -> None: ··· 111 105 fn: Callable[[list[dict[str, Any]]], list[dict[str, Any]]], 112 106 ) -> list[dict[str, Any]]: 113 107 """Apply a locked read-modify-write cycle to review-candidates.jsonl.""" 114 - review_candidates_dir() 115 - lock_path = review_candidates_lock_path() 116 - # Lock file contents are irrelevant; opening with "w" matches the existing pattern. 117 - with open(lock_path, "w", encoding="utf-8") as lock_file: 118 - fcntl.flock(lock_file, fcntl.LOCK_EX) 119 - try: 120 - rows = load_candidates() 121 - new_rows = fn(rows) 122 - save_candidates(new_rows) 123 - return new_rows 124 - finally: 125 - fcntl.flock(lock_file, fcntl.LOCK_UN) 108 + with hold_lock(review_candidates_path()): 109 + rows = load_candidates() 110 + new_rows = fn(rows) 111 + save_candidates(new_rows) 112 + return new_rows 126 113 127 114 128 115 def utc_now_iso() -> str:
+10 -16
solstone/think/entities/saving.py
··· 8 8 - save_detected_entity: Concurrency-safe single entity detection with file locking 9 9 """ 10 10 11 - import fcntl 12 11 import json 13 12 import random 14 13 import time 15 14 16 - from solstone.think.entities.core import EntityDict, atomic_write, entity_slug 15 + from solstone.think.entities.core import EntityDict, entity_slug 17 16 from solstone.think.entities.journal import ( 18 17 create_journal_entity, 19 18 load_journal_entity, ··· 25 24 load_entities, 26 25 ) 27 26 from solstone.think.entities.relationships import save_facet_relationship 27 + from solstone.think.journal_io import atomic_replace, hold_lock 28 28 29 29 30 30 def _save_entities_detected(facet: str, entities: list[EntityDict], day: str) -> None: ··· 45 45 46 46 # Format as JSONL and write atomically 47 47 content = "".join(json.dumps(e, ensure_ascii=False) + "\n" for e in sorted_entities) 48 - atomic_write(path, content, prefix="entities_") 48 + atomic_replace(path, content) 49 49 clear_entity_loading_cache() 50 50 51 51 ··· 208 208 OSError: If all retries exhausted on transient errors 209 209 """ 210 210 path = detected_entities_path(facet, day) 211 - lock_path = path.parent / f"{path.name}.lock" 212 211 213 212 last_error: Exception | None = None 214 213 for attempt in range(max_retries): 215 214 try: 216 - path.parent.mkdir(parents=True, exist_ok=True) 217 - with open(lock_path, "w") as lock_file: 218 - fcntl.flock(lock_file, fcntl.LOCK_EX) 219 - try: 220 - # Fresh load inside lock — sees all prior writers' changes 221 - clear_entity_loading_cache() 222 - entities = load_entities(facet, day) 223 - entities = modify_fn(entities) 224 - _save_entities_detected(facet, entities, day) 225 - return entities 226 - finally: 227 - fcntl.flock(lock_file, fcntl.LOCK_UN) 215 + with hold_lock(path): 216 + # Fresh load inside lock — sees all prior writers' changes 217 + clear_entity_loading_cache() 218 + entities = load_entities(facet, day) 219 + entities = modify_fn(entities) 220 + _save_entities_detected(facet, entities, day) 221 + return entities 228 222 except ValueError: 229 223 raise # Logical errors (duplicate, not found) — don't retry 230 224 except OSError as exc:
+15 -1
solstone/think/entities/seeding.py
··· 9 9 10 10 from __future__ import annotations 11 11 12 + import logging 12 13 from typing import TYPE_CHECKING 14 + 15 + from solstone.think.journal_io import LockTimeout 13 16 14 17 if TYPE_CHECKING: 15 18 from solstone.think.entities.core import EntityDict 19 + 20 + logger = logging.getLogger(__name__) 16 21 17 22 18 23 def seed_entities( ··· 109 114 110 115 ensure_facet(facet) 111 116 facet_ensured = True 112 - add_observation(facet, resolved_name, obs_content, source_day=day) 117 + try: 118 + add_observation( 119 + facet, resolved_name, obs_content, source_day=day 120 + ) 121 + except LockTimeout: 122 + logger.warning( 123 + "observations busy for %s; skipping remaining", 124 + resolved_name, 125 + ) 126 + break 113 127 existing_contents.add(obs_content) 114 128 115 129 return resolved
+35 -7
tests/test_entities.py
··· 409 409 """Test that save_detected_entity retries on transient OSError.""" 410 410 from unittest.mock import patch 411 411 412 + from solstone.think.journal_io import atomic_replace as _real_atomic_replace 413 + 412 414 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 413 415 (tmp_path / "facets" / "test_facet" / "entities").mkdir(parents=True) 414 416 415 417 call_count = 0 416 - original_atomic_write = __import__( 417 - "solstone.think.entities.core", fromlist=["atomic_write"] 418 - ).atomic_write 419 418 420 - def flaky_atomic_write(path, content, prefix=".tmp_"): 419 + def flaky_atomic_replace(path, data, *, mode=None): 421 420 nonlocal call_count 422 421 call_count += 1 423 422 if call_count == 1: 424 423 raise PermissionError("Simulated transient error") 425 - return original_atomic_write(path, content, prefix) 424 + return _real_atomic_replace(path, data, mode=mode) 426 425 427 426 with patch( 428 - "solstone.think.entities.saving.atomic_write", side_effect=flaky_atomic_write 427 + "solstone.think.entities.saving.atomic_replace", 428 + side_effect=flaky_atomic_replace, 429 429 ): 430 430 save_detected_entity("test_facet", "20250101", "Person", "Alice", "Friend") 431 431 432 - assert call_count == 2 # First attempt failed, second succeeded 432 + assert call_count == 2 433 433 loaded = load_entities("test_facet", "20250101") 434 434 assert len(loaded) == 1 435 435 assert loaded[0]["name"] == "Alice" ··· 2072 2072 # Verify persistence 2073 2073 loaded = load_observations("personal", "Alice") 2074 2074 assert len(loaded) == 2 2075 + 2076 + 2077 + def test_add_observation_retry_on_error(fixture_journal, tmp_path, monkeypatch): 2078 + """Test that add_observation retries on transient OSError.""" 2079 + from unittest.mock import patch 2080 + 2081 + from solstone.think.journal_io import atomic_replace as _real_atomic_replace 2082 + 2083 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 2084 + 2085 + call_count = 0 2086 + 2087 + def flaky_atomic_replace(path, data, *, mode=None): 2088 + nonlocal call_count 2089 + call_count += 1 2090 + if call_count == 1: 2091 + raise PermissionError("Simulated transient error") 2092 + return _real_atomic_replace(path, data, mode=mode) 2093 + 2094 + with patch( 2095 + "solstone.think.entities.observations.atomic_replace", 2096 + side_effect=flaky_atomic_replace, 2097 + ): 2098 + add_observation("personal", "Alice", "Prefers async communication", "20250113") 2099 + 2100 + assert call_count == 2 2101 + loaded = load_observations("personal", "Alice") 2102 + assert [obs["content"] for obs in loaded] == ["Prefers async communication"] 2075 2103 2076 2104 2077 2105 def test_add_observation_empty_content(fixture_journal, tmp_path, monkeypatch):
+164
tests/test_entities_locking.py
··· 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 + # Copyright (c) 2026 sol pbc 3 + 4 + from __future__ import annotations 5 + 6 + import multiprocessing 7 + import os 8 + import traceback 9 + from pathlib import Path 10 + from queue import Empty 11 + from typing import Any 12 + 13 + 14 + def _observation_worker( 15 + journal_path: str, 16 + barrier: Any, 17 + errors: Any, 18 + index: int, 19 + ) -> None: 20 + os.environ["SOLSTONE_JOURNAL"] = journal_path 21 + try: 22 + from solstone.think.entities.observations import add_observation 23 + 24 + barrier.wait(timeout=5) 25 + add_observation("work", "Alice", f"obs-{index}", source_day="20250101") 26 + except BaseException: 27 + errors.put(traceback.format_exc()) 28 + raise 29 + 30 + 31 + def _detected_entity_worker( 32 + journal_path: str, 33 + barrier: Any, 34 + errors: Any, 35 + index: int, 36 + ) -> None: 37 + os.environ["SOLSTONE_JOURNAL"] = journal_path 38 + try: 39 + from solstone.think.entities.saving import save_detected_entity 40 + 41 + barrier.wait(timeout=5) 42 + save_detected_entity("work", "20250101", "Person", f"E{index}", "d") 43 + except BaseException: 44 + errors.put(traceback.format_exc()) 45 + raise 46 + 47 + 48 + def _candidate_worker( 49 + journal_path: str, 50 + barrier: Any, 51 + errors: Any, 52 + index: int, 53 + ) -> None: 54 + os.environ["SOLSTONE_JOURNAL"] = journal_path 55 + try: 56 + from solstone.think.entities.review_candidates import locked_modify_candidates 57 + 58 + def mutate(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: 59 + next_rows = list(rows) 60 + next_rows.append( 61 + { 62 + "facet": "work", 63 + "source_slug": f"s{index}", 64 + "target_slug": "target", 65 + } 66 + ) 67 + return next_rows 68 + 69 + barrier.wait(timeout=5) 70 + locked_modify_candidates(mutate) 71 + except BaseException: 72 + errors.put(traceback.format_exc()) 73 + raise 74 + 75 + 76 + def _drain_errors(errors: Any) -> list[str]: 77 + found = [] 78 + while True: 79 + try: 80 + found.append(errors.get_nowait()) 81 + except Empty: 82 + return found 83 + 84 + 85 + def _join_processes(processes: list[Any], errors: Any) -> None: 86 + for process in processes: 87 + process.start() 88 + for process in processes: 89 + process.join(timeout=10) 90 + for process in processes: 91 + if process.is_alive(): 92 + process.terminate() 93 + process.join(timeout=2) 94 + 95 + error_text = "\n".join(_drain_errors(errors)) 96 + assert all(not process.is_alive() for process in processes), error_text 97 + assert all(process.exitcode == 0 for process in processes), error_text 98 + 99 + 100 + def _run_workers(tmp_path: Path, monkeypatch: Any, name: str, target: Any) -> Path: 101 + ctx = multiprocessing.get_context("spawn") 102 + journal = tmp_path / name 103 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal)) 104 + barrier = ctx.Barrier(4) 105 + errors = ctx.Queue() 106 + processes = [ 107 + ctx.Process(target=target, args=(str(journal), barrier, errors, i)) 108 + for i in range(4) 109 + ] 110 + 111 + _join_processes(processes, errors) 112 + return journal 113 + 114 + 115 + def test_add_observation_serializes_process_writers( 116 + tmp_path: Path, 117 + monkeypatch: Any, 118 + ) -> None: 119 + _run_workers(tmp_path, monkeypatch, "observations", _observation_worker) 120 + 121 + from solstone.think.entities.observations import ( 122 + clear_observation_cache, 123 + load_observations, 124 + ) 125 + 126 + clear_observation_cache() 127 + observations = load_observations("work", "Alice") 128 + 129 + assert sorted(obs["content"] for obs in observations) == [ 130 + "obs-0", 131 + "obs-1", 132 + "obs-2", 133 + "obs-3", 134 + ] 135 + 136 + 137 + def test_save_detected_entity_serializes_process_writers( 138 + tmp_path: Path, 139 + monkeypatch: Any, 140 + ) -> None: 141 + _run_workers(tmp_path, monkeypatch, "detected", _detected_entity_worker) 142 + 143 + from solstone.think.entities.loading import ( 144 + clear_entity_loading_cache, 145 + load_entities, 146 + ) 147 + 148 + clear_entity_loading_cache() 149 + entities = load_entities("work", "20250101") 150 + 151 + assert sorted(entity["name"] for entity in entities) == ["E0", "E1", "E2", "E3"] 152 + 153 + 154 + def test_locked_modify_candidates_serializes_process_writers( 155 + tmp_path: Path, 156 + monkeypatch: Any, 157 + ) -> None: 158 + _run_workers(tmp_path, monkeypatch, "candidates", _candidate_worker) 159 + 160 + from solstone.think.entities.review_candidates import load_candidates 161 + 162 + rows = load_candidates() 163 + 164 + assert sorted(row["source_slug"] for row in rows) == ["s0", "s1", "s2", "s3"]
+15 -5
tests/test_think_review_candidates.py
··· 6 6 import logging 7 7 import threading 8 8 from pathlib import Path 9 + from unittest.mock import patch 9 10 10 11 import pytest 11 12 ··· 17 18 load_candidates, 18 19 locked_modify_candidates, 19 20 review_candidates_dir, 20 - review_candidates_lock_path, 21 21 review_candidates_path, 22 22 save_candidates, 23 23 touch_updated, ··· 43 43 assert ( 44 44 review_candidates_path() 45 45 == candidate_journal / "entities" / "review-candidates.jsonl" 46 - ) 47 - assert ( 48 - review_candidates_lock_path() 49 - == candidate_journal / "entities" / ".review-candidates.lock" 50 46 ) 51 47 52 48 ··· 201 197 assert load_candidates() == [ 202 198 {"facet": "work", "source_slug": "s", "target_slug": "t"} 203 199 ] 200 + 201 + 202 + def test_locked_modify_candidates_does_not_retry_on_write_error(candidate_journal): 203 + def mutate(rows): 204 + return list(rows) + [{"facet": "work", "source_slug": "s", "target_slug": "t"}] 205 + 206 + with patch( 207 + "solstone.think.entities.review_candidates.atomic_replace", 208 + side_effect=PermissionError("Simulated write error"), 209 + ) as atomic_replace: 210 + with pytest.raises(OSError): 211 + locked_modify_candidates(mutate) 212 + 213 + assert atomic_replace.call_count == 1 204 214 205 215 206 216 def test_locked_modify_candidates_serializes_threads(candidate_journal):