solpbc.org / solstone
personal memory agent
0
Fork 0

Configure Feed

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

refactor(facets): single-source facet/convey-config I/O on journal_io primitives

Route the facet domain's file-write mechanics onto the shared
solstone/think/journal_io primitives and close two durability/concurrency
races on config/convey.json. On-disk formats are byte-identical
(ensure_ascii=False preserved throughout — the default facet emoji is
non-ASCII), so this is write-mechanics + locking only.

- facets.py: collapse the three facet.json atomic-writer copies into one
_write_facet_json on atomic_replace; route the action-log append through
append_text; make delete_facet/rename_facet's convey.json read-modify-write
atomic and locked under hold_lock.
- convey/config.py: replace save_convey_config with a locked
locked_modify_convey_config RMW helper (unlocked inner _write_convey_config
to avoid self-deadlock) backing all six config routes + set_selected_facet;
the CLI (facets.py) and web server now share the same config/convey.json.lock,
closing the lost-update window. set_selected_facet stays best-effort
(template-context callers must not fault on contention).
- facet_review_candidates.py: locked_modify_candidates onto hold_lock,
_save_jsonl_rows onto atomic_replace; drop the cross-domain
entities.core.atomic_write import, fcntl, and the dead lock-path helper
(sidecar is now review-candidates.jsonl.lock).
- Add CONVEY_BUSY (503); convey config routes translate LockTimeout to it,
curation facet-candidate handlers reuse ENTITY_BUSY.
- Register convey/config.py as a journal_io owner and add the config/convey.json
row to the AGENTS.md L2 ownership table.
- Tests: multiprocess cross-writer + failure-injection coverage for convey.json,
review-candidates, and facet.json.

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:55 PM -0600) commit a2d4ce20 parent 1a2c5f64
+526 -241
+1
AGENTS.md
··· 190 190 | Awareness (`awareness/current.json`) | `solstone/think/awareness.py` | 191 191 | Todos (`facets/*/todos/*.jsonl`) | `solstone/apps/todos/todo.py` + `solstone/apps/todos/call.py` | 192 192 | Config (`config/journal.json`) | `solstone/think/journal_config.py` | 193 + | Convey config (`config/convey.json`) | `solstone/convey/config.py` + `solstone/think/facets.py` | 193 194 | Chat config (`config/chat.json`) | `solstone/apps/chat/config.py` | 194 195 | Vertex credentials (`.config/vertex-credentials.json`) | `solstone/apps/settings/vertex_credentials.py` | 195 196 | Speaker labels (`chronicle/**/talents/speaker_labels.json`) | `solstone/apps/speakers/attribution.py` |
+1
scripts/check_journal_io_access.py
··· 81 81 { 82 82 "solstone/apps/entities/call.py", 83 83 "solstone/apps/chat/config.py", 84 + "solstone/convey/config.py", 84 85 "solstone/apps/speakers/attribution.py", 85 86 "solstone/apps/speakers/owner.py", 86 87 "solstone/apps/speakers/routes.py",
+8 -2
solstone/apps/curation/routes.py
··· 74 74 name_key = str(_required(data, "name_key")) 75 75 except KeyError: 76 76 return _missing_field("name_key") 77 - return _result_response(accept_facet_candidate(name_key)) 77 + try: 78 + return _result_response(accept_facet_candidate(name_key)) 79 + except LockTimeout: 80 + return error_response(ENTITY_BUSY, detail="suggestions are busy; try again") 78 81 79 82 80 83 @curation_bp.post("/api/facet/dismiss") ··· 84 87 name_key = str(_required(data, "name_key")) 85 88 except KeyError: 86 89 return _missing_field("name_key") 87 - return _result_response(dismiss_facet_candidate(name_key)) 90 + try: 91 + return _result_response(dismiss_facet_candidate(name_key)) 92 + except LockTimeout: 93 + return error_response(ENTITY_BUSY, detail="suggestions are busy; try again") 88 94 89 95 90 96 @curation_bp.post("/api/entity/preview")
+8 -9
solstone/apps/settings/maint/003_seed_default_app_navigation.py
··· 7 7 8 8 import logging 9 9 import sys 10 + from typing import Any 10 11 11 12 import solstone.convey.state as convey_state 12 13 from solstone.convey.config import ( 13 - load_convey_config, 14 - save_convey_config, 14 + locked_modify_convey_config, 15 15 seed_default_app_navigation, 16 16 ) 17 17 from solstone.think.utils import get_journal ··· 36 36 37 37 convey_state.journal_root = str(journal) 38 38 39 - config = load_convey_config() 40 - if not seed_default_app_navigation(config): 41 - print("Default app navigation already present.") 42 - return 39 + def _seed(config: dict[str, Any]) -> dict[str, Any] | None: 40 + return config if seed_default_app_navigation(config) else None 43 41 44 42 try: 45 - saved = save_convey_config(config) 43 + result = locked_modify_convey_config(_seed) 46 44 except Exception as exc: 47 45 _fail("default app navigation seed convey-config PERSIST failed", exc) 48 46 49 - if not saved: 50 - _fail("default app navigation seed convey-config PERSIST failed") 47 + if result is None: 48 + print("Default app navigation already present.") 49 + return 51 50 52 51 print("Seeded default app navigation.") 53 52
+85 -102
solstone/convey/config.py
··· 5 5 6 6 from __future__ import annotations 7 7 8 + import json 8 9 import logging 10 + from collections.abc import Callable 9 11 from pathlib import Path 10 12 from typing import Any 11 13 12 14 from flask import Blueprint, request 15 + 16 + from solstone.think.journal_io import LockTimeout, atomic_replace, hold_lock 13 17 14 18 from . import state 15 19 from .reasons import ( 20 + CONVEY_BUSY, 16 21 CONVEY_OPERATION_FAILED, 17 22 INVALID_CONFIG_VALUE, 18 23 INVALID_JSON_REQUEST, 19 24 MISSING_REQUIRED_FIELD, 20 25 ) 21 - from .utils import error_response, load_json, save_json, success_response 26 + from .utils import error_response, load_json, success_response 22 27 23 28 logger = logging.getLogger(__name__) 24 29 ··· 69 74 return config.get("reporting", {}).get("enabled", True) 70 75 71 76 72 - def save_convey_config(config: dict[str, Any]) -> bool: 73 - """Save config/convey.json atomically. 77 + def _write_convey_config(config: dict[str, Any]) -> None: 78 + """Atomically persist convey config. Caller MUST hold the convey.json lock.""" 79 + atomic_replace( 80 + _get_config_path(), 81 + json.dumps(config, indent=2, ensure_ascii=False) + "\n", 82 + ) 74 83 75 - Args: 76 - config: Configuration dict to save 77 84 78 - Returns: 79 - True if successful, False otherwise 80 - """ 81 - config_path = _get_config_path() 85 + def locked_modify_convey_config( 86 + transform: Callable[[dict[str, Any]], dict[str, Any] | None], 87 + ) -> dict[str, Any] | None: 88 + """Apply a locked read-modify-write to config/convey.json. 82 89 83 - # Ensure config directory exists 84 - config_path.parent.mkdir(parents=True, exist_ok=True) 85 - 86 - return save_json(config_path, config, indent=2) 90 + Reads the current config under an exclusive lock, applies ``transform``, 91 + and atomically persists the result. If ``transform`` returns ``None`` the 92 + write is skipped (no-op). Returns the persisted config, or ``None`` when 93 + skipped. 94 + """ 95 + path = _get_config_path() 96 + with hold_lock(path): 97 + config = load_convey_config() 98 + new_config = transform(config) 99 + if new_config is None: 100 + return None 101 + _write_convey_config(new_config) 102 + return new_config 87 103 88 104 89 105 def seed_default_app_navigation(config: dict[str, Any]) -> bool: ··· 117 133 118 134 119 135 def set_selected_facet(facet: str | None) -> None: 120 - """Update selected facet in config. 121 - 122 - Args: 123 - facet: Facet name to select, or None to clear selection 124 - """ 125 - config = load_convey_config() 126 - 127 - # Ensure facets section exists 128 - if "facets" not in config: 129 - config["facets"] = {} 136 + """Update selected facet in config (best-effort; never raises).""" 130 137 131 - # Update selected field 132 - config["facets"]["selected"] = facet 138 + def _transform(config: dict[str, Any]) -> dict[str, Any]: 139 + config.setdefault("facets", {})["selected"] = facet 140 + return config 133 141 134 - # Save config (async safe - doesn't block if write fails) 135 - success = save_convey_config(config) 136 - if not success: 137 - logger.warning(f"Failed to save selected facet: {facet}") 142 + try: 143 + locked_modify_convey_config(_transform) 144 + except (LockTimeout, OSError) as exc: 145 + logger.warning("Failed to save selected facet %s: %s", facet, exc) 138 146 139 147 140 148 def apply_facet_order(facets: list[dict], config: dict) -> list[dict]: ··· 350 358 detail=f"Invalid config: {error_msg}", 351 359 ) 352 360 353 - # Merge with existing config (partial updates supported) 354 - current_config = load_convey_config() 355 - 356 - # Deep merge facets section 357 - if "facets" in new_config: 358 - if "facets" not in current_config: 359 - current_config["facets"] = {} 360 - current_config["facets"].update(new_config["facets"]) 361 + def _transform(config: dict[str, Any]) -> dict[str, Any]: 362 + if "facets" in new_config: 363 + config.setdefault("facets", {}).update(new_config["facets"]) 364 + if "apps" in new_config: 365 + config.setdefault("apps", {}).update(new_config["apps"]) 366 + if "reporting" in new_config: 367 + config.setdefault("reporting", {}).update(new_config["reporting"]) 368 + return config 361 369 362 - # Deep merge apps section 363 - if "apps" in new_config: 364 - if "apps" not in current_config: 365 - current_config["apps"] = {} 366 - current_config["apps"].update(new_config["apps"]) 370 + persisted = locked_modify_convey_config(_transform) 371 + return success_response({"config": persisted}) 367 372 368 - # Deep merge reporting section 369 - if "reporting" in new_config: 370 - if "reporting" not in current_config: 371 - current_config["reporting"] = {} 372 - current_config["reporting"].update(new_config["reporting"]) 373 - 374 - # Save updated config 375 - success = save_convey_config(current_config) 376 - if not success: 377 - return error_response( 378 - CONVEY_OPERATION_FAILED, 379 - detail="Failed to save configuration", 380 - ) 381 - 382 - return success_response({"config": current_config}) 383 - 373 + except LockTimeout: 374 + return error_response( 375 + CONVEY_BUSY, 376 + detail="Interface settings are busy; try again", 377 + ) 384 378 except Exception as e: 385 379 logger.error(f"Failed to update config: {e}", exc_info=True) 386 380 return error_response( ··· 418 412 detail="'order' must contain only strings", 419 413 ) 420 414 421 - # Load config and update facets.order 422 - config = load_convey_config() 423 - if "facets" not in config: 424 - config["facets"] = {} 425 - config["facets"]["order"] = order 415 + def _transform(config: dict[str, Any]) -> dict[str, Any]: 416 + config.setdefault("facets", {})["order"] = order 417 + return config 426 418 427 - # Save 428 - success = save_convey_config(config) 429 - if not success: 430 - return error_response( 431 - CONVEY_OPERATION_FAILED, 432 - detail="Failed to save facet order", 433 - ) 419 + locked_modify_convey_config(_transform) 434 420 435 421 return success_response({"order": order}) 436 422 423 + except LockTimeout: 424 + return error_response( 425 + CONVEY_BUSY, 426 + detail="Interface settings are busy; try again", 427 + ) 437 428 except Exception as e: 438 429 logger.error(f"Failed to update facet order: {e}", exc_info=True) 439 430 return error_response( ··· 471 462 detail="'order' must contain only strings", 472 463 ) 473 464 474 - # Load config and update apps.order 475 - config = load_convey_config() 476 - if "apps" not in config: 477 - config["apps"] = {} 478 - config["apps"]["order"] = order 465 + def _transform(config: dict[str, Any]) -> dict[str, Any]: 466 + config.setdefault("apps", {})["order"] = order 467 + return config 479 468 480 - # Save 481 - success = save_convey_config(config) 482 - if not success: 483 - return error_response( 484 - CONVEY_OPERATION_FAILED, 485 - detail="Failed to save app order", 486 - ) 469 + locked_modify_convey_config(_transform) 487 470 488 471 return success_response({"order": order}) 489 472 473 + except LockTimeout: 474 + return error_response( 475 + CONVEY_BUSY, 476 + detail="Interface settings are busy; try again", 477 + ) 490 478 except Exception as e: 491 479 logger.error(f"Failed to update app order: {e}", exc_info=True) 492 480 return error_response( ··· 524 512 detail="'starred' must be a boolean", 525 513 ) 526 514 527 - # Load config and update apps.starred 528 - config = load_convey_config() 529 - if "apps" not in config: 530 - config["apps"] = {} 515 + def _transform(config: dict[str, Any]) -> dict[str, Any]: 516 + apps_config = config.setdefault("apps", {}) 517 + starred_apps = set(apps_config.get("starred", [])) 518 + if starred: 519 + starred_apps.add(app_name) 520 + else: 521 + starred_apps.discard(app_name) 522 + apps_config["starred"] = sorted(starred_apps) 523 + return config 531 524 532 - starred_apps = set(config["apps"].get("starred", [])) 533 - 534 - if starred: 535 - starred_apps.add(app_name) 536 - else: 537 - starred_apps.discard(app_name) 538 - 539 - config["apps"]["starred"] = sorted(starred_apps) 540 - 541 - # Save 542 - success = save_convey_config(config) 543 - if not success: 544 - return error_response( 545 - CONVEY_OPERATION_FAILED, 546 - detail="Failed to save app starred status", 547 - ) 525 + locked_modify_convey_config(_transform) 548 526 549 527 return success_response({"app": app_name, "starred": starred}) 550 528 529 + except LockTimeout: 530 + return error_response( 531 + CONVEY_BUSY, 532 + detail="Interface settings are busy; try again", 533 + ) 551 534 except Exception as e: 552 535 logger.error(f"Failed to toggle app star: {e}", exc_info=True) 553 536 return error_response(
+5
solstone/convey/reasons.py
··· 110 110 "I couldn't update the interface settings.", 111 111 500, 112 112 ) 113 + CONVEY_BUSY = Reason( 114 + "convey_busy", 115 + "I couldn't update the interface settings right now because they were busy. Try again in a moment.", 116 + 503, 117 + ) 113 118 NETWORK_SECURITY_REQUIRES_PASSWORD = Reason( 114 119 "network_security_requires_password", 115 120 "I couldn't change network access until a password is set.",
+7 -4
solstone/convey/root.py
··· 41 41 42 42 from . import bridge as convey_bridge 43 43 from .config import ( 44 - load_convey_config, 45 - save_convey_config, 44 + locked_modify_convey_config, 46 45 seed_default_app_navigation, 47 46 ) 48 47 from .copy import LOGIN_NO_PASSWORD_CONFIGURED ··· 369 368 370 369 write_journal_config(config) 371 370 372 - config = load_convey_config() 373 - if seed_default_app_navigation(config) and not save_convey_config(config): 371 + def _seed(config: dict[str, Any]) -> dict[str, Any] | None: 372 + return config if seed_default_app_navigation(config) else None 373 + 374 + try: 375 + locked_modify_convey_config(_seed) 376 + except Exception: 374 377 logger.error("default app navigation seed convey-config PERSIST failed") 375 378 376 379 session["logged_in"] = True
+12 -23
solstone/think/facet_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__) ··· 31 30 def facet_review_candidates_path() -> Path: 32 31 """Return the facet review-candidates JSONL path.""" 33 32 return facet_review_candidates_dir() / "review-candidates.jsonl" 34 - 35 - 36 - def facet_review_candidates_lock_path() -> Path: 37 - """Return the sibling lock path for review-candidates.jsonl.""" 38 - return facet_review_candidates_dir() / ".review-candidates.lock" 39 33 40 34 41 35 def _load_jsonl_rows(path: Path) -> list[dict[str, Any]]: ··· 77 71 78 72 def _save_jsonl_rows(path: Path, rows: list[dict[str, Any]]) -> None: 79 73 """Write *rows* to *path* as JSONL using an atomic replace.""" 80 - content = "" 81 - if rows: 82 - content = "\n".join(json.dumps(row, ensure_ascii=False) for row in rows) + "\n" 83 - atomic_write(path, content) 74 + content = ( 75 + "\n".join(json.dumps(row, ensure_ascii=False) for row in rows) + "\n" 76 + if rows 77 + else "" 78 + ) 79 + atomic_replace(path, content) 84 80 85 81 86 82 def save_candidates(rows: list[dict[str, Any]]) -> None: ··· 107 103 fn: Callable[[list[dict[str, Any]]], list[dict[str, Any]]], 108 104 ) -> list[dict[str, Any]]: 109 105 """Apply a locked read-modify-write cycle to review-candidates.jsonl.""" 110 - facet_review_candidates_dir() 111 - lock_path = facet_review_candidates_lock_path() 112 - # Lock file contents are irrelevant; opening with "w" matches the existing pattern. 113 - with open(lock_path, "w", encoding="utf-8") as lock_file: 114 - fcntl.flock(lock_file, fcntl.LOCK_EX) 115 - try: 116 - rows = load_candidates() 117 - new_rows = fn(rows) 118 - save_candidates(new_rows) 119 - return new_rows 120 - finally: 121 - fcntl.flock(lock_file, fcntl.LOCK_UN) 106 + with hold_lock(facet_review_candidates_path()): 107 + rows = load_candidates() 108 + new_rows = fn(rows) 109 + save_candidates(new_rows) 110 + return new_rows 122 111 123 112 124 113 def utc_now_iso() -> str:
+52 -95
solstone/think/facets.py
··· 13 13 from typing import Any, Optional 14 14 15 15 from solstone.think.entities import get_identity_names 16 + from solstone.think.journal_io import append_text, atomic_replace, hold_lock 16 17 from solstone.think.utils import day_dirs, day_path, get_journal, iter_segments 17 18 18 19 ··· 185 186 else: 186 187 log_path = Path(journal) / "config" / "actions" / f"{day}.jsonl" 187 188 188 - # Ensure parent directory exists 189 - log_path.parent.mkdir(parents=True, exist_ok=True) 190 - 191 189 # Create log entry 192 190 entry = { 193 191 "timestamp": datetime.now(timezone.utc).isoformat(), ··· 206 204 entry["use_id"] = use_id 207 205 208 206 # Append to log file 209 - with open(log_path, "a", encoding="utf-8") as f: 210 - f.write(json.dumps(entry, ensure_ascii=False) + "\n") 207 + append_text(log_path, json.dumps(entry, ensure_ascii=False)) 211 208 212 209 213 210 def log_call_action( ··· 244 241 245 242 def _write_facet_json(path: Path, data: dict[str, Any]) -> None: 246 243 """Write facet metadata atomically.""" 247 - import tempfile 248 - 249 - temp_fd, temp_path = tempfile.mkstemp(dir=path.parent, suffix=".json", text=True) 250 - try: 251 - with os.fdopen(temp_fd, "w", encoding="utf-8") as f: 252 - json.dump(data, f, indent=2, ensure_ascii=False) 253 - f.write("\n") 254 - os.replace(temp_path, path) 255 - except Exception: 256 - try: 257 - os.unlink(temp_path) 258 - except Exception: 259 - pass 260 - raise 244 + atomic_replace(path, json.dumps(data, indent=2, ensure_ascii=False) + "\n") 261 245 262 246 263 247 def ensure_facet(slug: str) -> bool: ··· 757 741 facet_data.pop("muted", None) 758 742 759 743 # Write back atomically 760 - import tempfile 761 - 762 - temp_fd, temp_path = tempfile.mkstemp( 763 - dir=facet_json_path.parent, suffix=".json", text=True 764 - ) 765 - try: 766 - with os.fdopen(temp_fd, "w", encoding="utf-8") as f: 767 - json.dump(facet_data, f, indent=2, ensure_ascii=False) 768 - f.write("\n") 769 - os.replace(temp_path, facet_json_path) 770 - except Exception: 771 - # Clean up temp file on error 772 - try: 773 - os.unlink(temp_path) 774 - except Exception: 775 - pass 776 - raise 744 + _write_facet_json(facet_json_path, facet_data) 777 745 778 746 # Log the change 779 747 action = "facet_mute" if muted else "facet_unmute" ··· 886 854 changed_fields[field] = {"old": old_value, "new": new_value} 887 855 facet_data[field] = new_value 888 856 889 - import tempfile 890 - 891 - temp_fd, temp_path = tempfile.mkstemp( 892 - dir=facet_json_path.parent, suffix=".json", text=True 893 - ) 894 - try: 895 - with os.fdopen(temp_fd, "w", encoding="utf-8") as f: 896 - json.dump(facet_data, f, indent=2, ensure_ascii=False) 897 - f.write("\n") 898 - os.replace(temp_path, facet_json_path) 899 - except Exception: 900 - try: 901 - os.unlink(temp_path) 902 - except Exception: 903 - pass 904 - raise 857 + _write_facet_json(facet_json_path, facet_data) 905 858 906 859 if changed_fields: 907 860 log_call_action( ··· 930 883 931 884 convey_config_path = Path(get_journal()) / "config" / "convey.json" 932 885 if convey_config_path.exists(): 933 - try: 934 - with open(convey_config_path, "r", encoding="utf-8") as f: 935 - config = json.load(f) 886 + with hold_lock(convey_config_path): 887 + try: 888 + with open(convey_config_path, "r", encoding="utf-8") as f: 889 + config = json.load(f) 936 890 937 - changed = False 938 - facets_config = config.get("facets", {}) 891 + changed = False 892 + facets_config = config.get("facets", {}) 939 893 940 - if facets_config.get("selected") == name: 941 - facets_config["selected"] = "" 942 - changed = True 894 + if facets_config.get("selected") == name: 895 + facets_config["selected"] = "" 896 + changed = True 943 897 944 - order = facets_config.get("order", []) 945 - if name in order: 946 - facets_config["order"] = [item for item in order if item != name] 947 - changed = True 898 + order = facets_config.get("order", []) 899 + if name in order: 900 + facets_config["order"] = [item for item in order if item != name] 901 + changed = True 948 902 949 - if changed: 950 - config["facets"] = facets_config 951 - with open(convey_config_path, "w", encoding="utf-8") as f: 952 - json.dump(config, f, indent=2, ensure_ascii=False) 953 - f.write("\n") 954 - except (json.JSONDecodeError, OSError): 955 - pass 903 + if changed: 904 + config["facets"] = facets_config 905 + atomic_replace( 906 + convey_config_path, 907 + json.dumps(config, indent=2, ensure_ascii=False) + "\n", 908 + ) 909 + except (json.JSONDecodeError, OSError): 910 + pass 956 911 957 912 log_params: dict = {"name": name} 958 913 if consent: ··· 1294 1249 # Step 2: Update config/convey.json 1295 1250 convey_config_path = Path(journal) / "config" / "convey.json" 1296 1251 if convey_config_path.exists(): 1297 - try: 1298 - with open(convey_config_path, "r", encoding="utf-8") as f: 1299 - config = json.load(f) 1252 + with hold_lock(convey_config_path): 1253 + try: 1254 + with open(convey_config_path, "r", encoding="utf-8") as f: 1255 + config = json.load(f) 1300 1256 1301 - changed = False 1302 - facets_config = config.get("facets", {}) 1257 + changed = False 1258 + facets_config = config.get("facets", {}) 1303 1259 1304 - if facets_config.get("selected") == old_name: 1305 - facets_config["selected"] = new_name 1306 - changed = True 1260 + if facets_config.get("selected") == old_name: 1261 + facets_config["selected"] = new_name 1262 + changed = True 1307 1263 1308 - order = facets_config.get("order", []) 1309 - if old_name in order: 1310 - facets_config["order"] = [ 1311 - new_name if name == old_name else name for name in order 1312 - ] 1313 - changed = True 1264 + order = facets_config.get("order", []) 1265 + if old_name in order: 1266 + facets_config["order"] = [ 1267 + new_name if name == old_name else name for name in order 1268 + ] 1269 + changed = True 1314 1270 1315 - if changed: 1316 - config["facets"] = facets_config 1317 - with open(convey_config_path, "w", encoding="utf-8") as f: 1318 - json.dump(config, f, indent=2, ensure_ascii=False) 1319 - f.write("\n") 1320 - print("Updated config/convey.json") 1321 - else: 1322 - print("No changes needed in config/convey.json") 1323 - except (json.JSONDecodeError, OSError) as exc: 1324 - logging.warning("Failed to update convey config: %s", exc) 1271 + if changed: 1272 + config["facets"] = facets_config 1273 + atomic_replace( 1274 + convey_config_path, 1275 + json.dumps(config, indent=2, ensure_ascii=False) + "\n", 1276 + ) 1277 + print("Updated config/convey.json") 1278 + else: 1279 + print("No changes needed in config/convey.json") 1280 + except (json.JSONDecodeError, OSError) as exc: 1281 + logging.warning("Failed to update convey config: %s", exc) 1325 1282 1326 1283 # Step 3: Advise index rebuild 1327 1284 print(
+5 -1
tests/test_convey_config.py
··· 145 145 from solstone.convey import root as root_module 146 146 147 147 (journal_copy / "config" / "convey.json").unlink() 148 - monkeypatch.setattr(root_module, "save_convey_config", lambda _config: False) 148 + 149 + def _fail_seed(_transform): 150 + raise OSError("simulated persist failure") 151 + 152 + monkeypatch.setattr(root_module, "locked_modify_convey_config", _fail_seed) 149 153 caplog.set_level(logging.ERROR, logger="solstone.convey.root") 150 154 app = create_app(str(journal_copy)) 151 155 app.config["TESTING"] = True
+186
tests/test_convey_config_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 json 7 + import multiprocessing 8 + import os 9 + import time 10 + import traceback 11 + from pathlib import Path 12 + from queue import Empty 13 + from typing import Any 14 + 15 + import pytest 16 + 17 + 18 + def _write_convey(path: Path, payload: dict[str, Any]) -> None: 19 + path.parent.mkdir(parents=True, exist_ok=True) 20 + path.write_text( 21 + json.dumps(payload, indent=2, ensure_ascii=False) + "\n", 22 + encoding="utf-8", 23 + ) 24 + 25 + 26 + def _seed_cross_writer_journal(journal: Path) -> Path: 27 + config_path = journal / "config" / "convey.json" 28 + _write_convey( 29 + config_path, 30 + { 31 + "facets": {"order": ["a", "b"], "selected": "a"}, 32 + "apps": {"order": ["x"]}, 33 + }, 34 + ) 35 + facet_dir = journal / "facets" / "b" 36 + facet_dir.mkdir(parents=True, exist_ok=True) 37 + (facet_dir / "facet.json").write_text( 38 + json.dumps( 39 + { 40 + "title": "B", 41 + "description": "", 42 + "color": "#667eea", 43 + "emoji": "📦", 44 + }, 45 + indent=2, 46 + ensure_ascii=False, 47 + ) 48 + + "\n", 49 + encoding="utf-8", 50 + ) 51 + return config_path 52 + 53 + 54 + def _delete_facet_worker( 55 + journal_path: str, 56 + barrier: Any, 57 + errors: Any, 58 + delay: float, 59 + ) -> None: 60 + os.environ["SOLSTONE_JOURNAL"] = journal_path 61 + try: 62 + barrier.wait(timeout=5) 63 + if delay: 64 + time.sleep(delay) 65 + 66 + from solstone.think.facets import delete_facet 67 + 68 + delete_facet("b") 69 + except BaseException: 70 + errors.put(traceback.format_exc()) 71 + raise 72 + 73 + 74 + def _convey_config_worker( 75 + journal_path: str, 76 + barrier: Any, 77 + errors: Any, 78 + delay: float, 79 + ) -> None: 80 + os.environ["SOLSTONE_JOURNAL"] = journal_path 81 + try: 82 + barrier.wait(timeout=5) 83 + if delay: 84 + time.sleep(delay) 85 + 86 + import solstone.convey.state as convey_state 87 + from solstone.convey.config import locked_modify_convey_config 88 + 89 + convey_state.journal_root = journal_path 90 + 91 + def _transform(config: dict[str, Any]) -> dict[str, Any]: 92 + config.setdefault("apps", {})["order"] = ["x", "y"] 93 + return config 94 + 95 + locked_modify_convey_config(_transform) 96 + except BaseException: 97 + errors.put(traceback.format_exc()) 98 + raise 99 + 100 + 101 + def _drain_errors(errors: Any) -> list[str]: 102 + found = [] 103 + while True: 104 + try: 105 + found.append(errors.get_nowait()) 106 + except Empty: 107 + return found 108 + 109 + 110 + def _join_processes(processes: list[Any], errors: Any) -> None: 111 + for process in processes: 112 + process.start() 113 + for process in processes: 114 + process.join(timeout=10) 115 + for process in processes: 116 + if process.is_alive(): 117 + process.terminate() 118 + process.join(timeout=2) 119 + 120 + error_text = "\n".join(_drain_errors(errors)) 121 + assert all(not process.is_alive() for process in processes), error_text 122 + assert all(process.exitcode == 0 for process in processes), error_text 123 + 124 + 125 + def test_convey_config_atomic_failure_preserves_existing_bytes( 126 + tmp_path: Path, 127 + monkeypatch: pytest.MonkeyPatch, 128 + ) -> None: 129 + journal = tmp_path / "journal" 130 + config_path = journal / "config" / "convey.json" 131 + original = { 132 + "facets": {"order": ["a"], "selected": "a"}, 133 + "apps": {"order": ["x"]}, 134 + } 135 + _write_convey(config_path, original) 136 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal)) 137 + 138 + import solstone.convey.state as convey_state 139 + from solstone.convey.config import locked_modify_convey_config 140 + 141 + convey_state.journal_root = str(journal) 142 + original_bytes = config_path.read_bytes() 143 + 144 + def fail_replace(_src: str, _dst: str) -> None: 145 + raise OSError("simulated replace failure") 146 + 147 + monkeypatch.setattr("solstone.think.journal_io.atomic.os.replace", fail_replace) 148 + 149 + with pytest.raises(OSError): 150 + locked_modify_convey_config(lambda config: {**config, "apps": {"order": ["z"]}}) 151 + 152 + assert config_path.read_bytes() == original_bytes 153 + assert list(config_path.parent.glob(".tmp_*")) == [] 154 + 155 + 156 + @pytest.mark.parametrize( 157 + ("delete_delay", "convey_delay"), 158 + [(0.0, 0.2), (0.2, 0.0)], 159 + ) 160 + def test_convey_config_lock_shared_with_think_facet_delete( 161 + tmp_path: Path, 162 + delete_delay: float, 163 + convey_delay: float, 164 + ) -> None: 165 + ctx = multiprocessing.get_context("spawn") 166 + journal = tmp_path / f"journal-{delete_delay}-{convey_delay}" 167 + config_path = _seed_cross_writer_journal(journal) 168 + barrier = ctx.Barrier(2) 169 + errors = ctx.Queue() 170 + processes = [ 171 + ctx.Process( 172 + target=_delete_facet_worker, 173 + args=(str(journal), barrier, errors, delete_delay), 174 + ), 175 + ctx.Process( 176 + target=_convey_config_worker, 177 + args=(str(journal), barrier, errors, convey_delay), 178 + ), 179 + ] 180 + 181 + _join_processes(processes, errors) 182 + 183 + data = json.loads(config_path.read_text(encoding="utf-8")) 184 + assert data["facets"]["order"] == ["a"] 185 + assert data["apps"]["order"] == ["x", "y"] 186 + assert not (journal / "facets" / "b").exists()
-5
tests/test_facet_review_candidates.py
··· 15 15 candidate_key, 16 16 dismiss_candidate, 17 17 facet_review_candidates_dir, 18 - facet_review_candidates_lock_path, 19 18 facet_review_candidates_path, 20 19 find_candidate, 21 20 load_candidates, ··· 45 44 assert ( 46 45 facet_review_candidates_path() 47 46 == candidate_journal / "facets" / "review-candidates.jsonl" 48 - ) 49 - assert ( 50 - facet_review_candidates_lock_path() 51 - == candidate_journal / "facets" / ".review-candidates.lock" 52 47 ) 53 48 54 49
+125
tests/test_facet_review_candidates_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 json 7 + import multiprocessing 8 + import os 9 + import traceback 10 + from pathlib import Path 11 + from queue import Empty 12 + from typing import Any 13 + 14 + import pytest 15 + 16 + from solstone.think.facet_review_candidates import ( 17 + facet_review_candidates_path, 18 + load_candidates, 19 + save_candidates, 20 + ) 21 + 22 + 23 + def _record_candidate_worker( 24 + journal_path: str, 25 + barrier: Any, 26 + errors: Any, 27 + index: int, 28 + ) -> None: 29 + os.environ["SOLSTONE_JOURNAL"] = journal_path 30 + try: 31 + barrier.wait(timeout=5) 32 + 33 + from solstone.think.facet_review_candidates import record_facet_candidate 34 + 35 + record_facet_candidate( 36 + name=f"Candidate {index}", 37 + name_key=f"candidate {index}", 38 + count=index + 1, 39 + window_days=14, 40 + samples=[ 41 + { 42 + "day": "20260602", 43 + "stream": "archon", 44 + "segment": f"09000{index}_300", 45 + } 46 + ], 47 + day="20260602", 48 + ) 49 + except BaseException: 50 + errors.put(traceback.format_exc()) 51 + raise 52 + 53 + 54 + def _drain_errors(errors: Any) -> list[str]: 55 + found = [] 56 + while True: 57 + try: 58 + found.append(errors.get_nowait()) 59 + except Empty: 60 + return found 61 + 62 + 63 + def _join_processes(processes: list[Any], errors: Any) -> None: 64 + for process in processes: 65 + process.start() 66 + for process in processes: 67 + process.join(timeout=10) 68 + for process in processes: 69 + if process.is_alive(): 70 + process.terminate() 71 + process.join(timeout=2) 72 + 73 + error_text = "\n".join(_drain_errors(errors)) 74 + assert all(not process.is_alive() for process in processes), error_text 75 + assert all(process.exitcode == 0 for process in processes), error_text 76 + 77 + 78 + def test_facet_review_candidate_atomic_failure_preserves_existing_bytes( 79 + tmp_path: Path, 80 + monkeypatch: pytest.MonkeyPatch, 81 + ) -> None: 82 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 83 + path = facet_review_candidates_path() 84 + seed = {"name": "Café", "name_key": "café", "status": "open"} 85 + original = json.dumps(seed, ensure_ascii=False) + "\n" 86 + path.write_text(original, encoding="utf-8") 87 + 88 + def fail_replace(_src: str, _dst: str) -> None: 89 + raise OSError("simulated replace failure") 90 + 91 + monkeypatch.setattr("solstone.think.journal_io.atomic.os.replace", fail_replace) 92 + 93 + with pytest.raises(OSError): 94 + save_candidates([{"name": "Changed", "name_key": "changed"}]) 95 + 96 + assert path.read_text(encoding="utf-8") == original 97 + assert list(path.parent.glob(".tmp_*")) == [] 98 + 99 + 100 + def test_facet_review_candidate_locked_modify_survives_multiprocess_writers( 101 + tmp_path: Path, 102 + monkeypatch: pytest.MonkeyPatch, 103 + ) -> None: 104 + ctx = multiprocessing.get_context("spawn") 105 + journal = tmp_path / "journal" 106 + barrier = ctx.Barrier(4) 107 + errors = ctx.Queue() 108 + processes = [ 109 + ctx.Process( 110 + target=_record_candidate_worker, 111 + args=(str(journal), barrier, errors, index), 112 + ) 113 + for index in range(4) 114 + ] 115 + 116 + _join_processes(processes, errors) 117 + 118 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal)) 119 + rows = load_candidates() 120 + assert sorted(row["name_key"] for row in rows) == [ 121 + "candidate 0", 122 + "candidate 1", 123 + "candidate 2", 124 + "candidate 3", 125 + ]
+31
tests/test_facets_durability.py
··· 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 + # Copyright (c) 2026 sol pbc 3 + 4 + from __future__ import annotations 5 + 6 + from pathlib import Path 7 + 8 + import pytest 9 + 10 + from solstone.think.facets import create_facet, update_facet 11 + 12 + 13 + def test_facet_json_atomic_failure_preserves_existing_bytes( 14 + tmp_path: Path, 15 + monkeypatch: pytest.MonkeyPatch, 16 + ) -> None: 17 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 18 + slug = create_facet("Home Reno") 19 + facet_json = tmp_path / "facets" / slug / "facet.json" 20 + original = facet_json.read_bytes() 21 + 22 + def fail_replace(_src: str, _dst: str) -> None: 23 + raise OSError("simulated replace failure") 24 + 25 + monkeypatch.setattr("solstone.think.journal_io.atomic.os.replace", fail_replace) 26 + 27 + with pytest.raises(OSError): 28 + update_facet(slug, title="New Home Reno") 29 + 30 + assert facet_json.read_bytes() == original 31 + assert list(facet_json.parent.glob(".tmp_*")) == []