From 42c243de1f074af6e1df8deffe6bc05db67bfc56 Mon Sep 17 00:00:00 2001 From: Ryan Duguid Date: Sat, 17 Oct 2026 18:34:08 +1100 Subject: [PATCH 1/2] fix: serialise persistent sleep state updates\n --- skillopt_sleep/__main__.py | 5 + skillopt_sleep/cycle.py | 14 +++ skillopt_sleep/state.py | 78 ++++++++++++--- tests/test_sleep_engine.py | 5 +- tests/test_sleep_state_concurrency.py | 139 ++++++++++++++++++++++++++ 5 files changed, 224 insertions(+), 17 deletions(-) create mode 100644 tests/test_sleep_state_concurrency.py diff --git a/skillopt_sleep/__main__.py b/skillopt_sleep/__main__.py index e3b7794c6..2482d5164 100644 --- a/skillopt_sleep/__main__.py +++ b/skillopt_sleep/__main__.py @@ -384,6 +384,11 @@ def _flush_handoff(backend, args) -> int: def _handoff_mine_and_pin(cfg, args, backend, snapshot: str, dry: bool): + with SleepState.lock(cfg.state_path): + return _handoff_mine_and_pin_locked(cfg, args, backend, snapshot, dry) + + +def _handoff_mine_and_pin_locked(cfg, args, backend, snapshot: str, dry: bool): """Harvest + mine with the same knobs as run_sleep_cycle (harvest window, target-skill filter, candidate-limit bump, LLM mining — routed through the handoff files like every other model call), then pin the result to diff --git a/skillopt_sleep/cycle.py b/skillopt_sleep/cycle.py index 4ca5dc135..0a4ca5d67 100644 --- a/skillopt_sleep/cycle.py +++ b/skillopt_sleep/cycle.py @@ -629,6 +629,20 @@ def run_sleep_cycle( dry_run: bool = False, clock: Optional[float] = None, backend: Optional[Backend] = None, +) -> CycleOutcome: + """Run one cycle with exclusive ownership of its shared state transaction.""" + cfg = cfg or load_config() + with SleepState.lock(cfg.state_path): + return _run_sleep_cycle(cfg, seed_tasks=seed_tasks, dry_run=dry_run, clock=clock, backend=backend) + + +def _run_sleep_cycle( + cfg: Optional[SleepConfig] = None, + *, + seed_tasks: Optional[List[TaskRecord]] = None, + dry_run: bool = False, + clock: Optional[float] = None, + backend: Optional[Backend] = None, ) -> CycleOutcome: """Run one full sleep cycle and return the outcome. diff --git a/skillopt_sleep/state.py b/skillopt_sleep/state.py index 267adedd5..84345f00b 100644 --- a/skillopt_sleep/state.py +++ b/skillopt_sleep/state.py @@ -14,7 +14,9 @@ import copy import json import os -from typing import Any, Dict, Optional +import tempfile +from contextlib import contextmanager +from typing import Any, Dict, Iterator, Optional def _now_iso(clock: Optional[float] = None) -> str: @@ -44,25 +46,69 @@ def __init__(self, path: str, data: Optional[Dict[str, Any]] = None) -> None: self.data = data if data is not None else copy.deepcopy(DEFAULT_STATE) # io --------------------------------------------------------------------- + @staticmethod + @contextmanager + def lock(path: str) -> Iterator[None]: + """Refuse overlapping writers before they read or start model work.""" + lock_path = os.path.realpath(path) + ".lock" + os.makedirs(os.path.dirname(lock_path), exist_ok=True) + descriptor = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600) + locked = False + try: + if os.name == "nt": + import msvcrt + if os.fstat(descriptor).st_size == 0: + os.write(descriptor, b"\0") + os.lseek(descriptor, 0, os.SEEK_SET) + try: + msvcrt.locking(descriptor, msvcrt.LK_NBLCK, 1) + except OSError as error: + raise RuntimeError("Sleep state is busy; retry after the active cycle finishes") from error + else: + import fcntl + try: + fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError as error: + raise RuntimeError("Sleep state is busy; retry after the active cycle finishes") from error + locked = True + yield + finally: + try: + if locked: + if os.name == "nt": + os.lseek(descriptor, 0, os.SEEK_SET) + msvcrt.locking(descriptor, msvcrt.LK_UNLCK, 1) + else: + fcntl.flock(descriptor, fcntl.LOCK_UN) + finally: + os.close(descriptor) + @classmethod def load(cls, path: str) -> "SleepState": - if os.path.exists(path): - try: - with open(path) as f: - data = json.load(f) - merged = copy.deepcopy(DEFAULT_STATE) - merged.update(data if isinstance(data, dict) else {}) - return cls(path, merged) - except Exception: - pass - return cls(path, copy.deepcopy(DEFAULT_STATE)) + try: + with open(path, encoding="utf-8") as f: + data = json.load(f) + except FileNotFoundError: + return cls(path, copy.deepcopy(DEFAULT_STATE)) + if not isinstance(data, dict): + raise ValueError("Sleep state must contain a JSON object") + merged = copy.deepcopy(DEFAULT_STATE) + merged.update(data) + return cls(path, merged) def save(self) -> None: - os.makedirs(os.path.dirname(self.path), exist_ok=True) - tmp = self.path + ".tmp" - with open(tmp, "w") as f: - json.dump(self.data, f, ensure_ascii=False, indent=2) - os.replace(tmp, self.path) + directory = os.path.dirname(self.path) or "." + os.makedirs(directory, exist_ok=True) + descriptor, tmp = tempfile.mkstemp(prefix=".sleep-state-", suffix=".tmp", dir=directory) + try: + with os.fdopen(descriptor, "w", encoding="utf-8") as f: + json.dump(self.data, f, ensure_ascii=False, indent=2) + os.replace(tmp, self.path) + finally: + try: + os.unlink(tmp) + except FileNotFoundError: + pass # accessors -------------------------------------------------------------- @property diff --git a/tests/test_sleep_engine.py b/tests/test_sleep_engine.py index 62fe982e7..0ed74773d 100644 --- a/tests/test_sleep_engine.py +++ b/tests/test_sleep_engine.py @@ -2569,7 +2569,10 @@ def test_tool_aware_cli_run_fails_without_writes_or_checkpoint(self): self.assertIn("Cursor tool-aware replay is temporarily disabled", stderr.getvalue()) run.assert_not_called() self.assertEqual(backend._cache, {}) - self.assertFalse(os.path.exists(os.path.join(tmp, ".skillopt-sleep"))) + self.assertEqual( + os.listdir(os.path.join(tmp, ".skillopt-sleep")), + ["state.json.lock"], + ) self.assertFalse(os.path.exists(os.path.join(project, ".skillopt-sleep"))) self.assertFalse(os.path.exists(target)) diff --git a/tests/test_sleep_state_concurrency.py b/tests/test_sleep_state_concurrency.py new file mode 100644 index 000000000..3dc0c08a7 --- /dev/null +++ b/tests/test_sleep_state_concurrency.py @@ -0,0 +1,139 @@ +"""Only one cycle may read/update/save a shared state at a time.""" +import multiprocessing +import os +from types import SimpleNamespace + +import pytest + +from skillopt_sleep import state as state_module +from skillopt_sleep.state import SleepState + + +def update_project(path, project, ready, release): + try: + with SleepState.lock(path): + state = SleepState.load(path) + ready.set() + assert release.wait(10) + state.set_last_harvest(project, "2026-01-01T00:00:00") + state.record_night({"project": project}) + state.save() + except RuntimeError: + raise SystemExit(2) + + +def exit_holding_lock(path): + with SleepState.lock(path): + os._exit(3) + + +def test_process_exit_releases_lock_without_changing_state(tmp_path): + path = tmp_path / "state.json" + original = b'{"night": 4}' + path.write_bytes(original) + process = multiprocessing.get_context("spawn").Process( + target=exit_holding_lock, args=(str(path),) + ) + try: + process.start() + process.join(10) + assert process.exitcode == 3 + with SleepState.lock(str(path)): + assert path.read_bytes() == original + finally: + if process.pid is not None: + if process.is_alive(): + process.terminate() + process.join(5) + + +def test_two_projects_refuse_overlap_then_preserve_both_updates(tmp_path): + context = multiprocessing.get_context("spawn") + path = str(tmp_path / "state.json") + ready, release = context.Event(), context.Event() + other_ready = context.Event() + first = context.Process(target=update_project, args=(path, "first", ready, release)) + children = [first] + try: + first.start() + assert ready.wait(10) + second = context.Process(target=update_project, args=(path, "second", other_ready, release)) + children.append(second) + second.start() + second.join(10) + assert second.exitcode == 2 + release.set() + first.join(10) + assert first.exitcode == 0 + retry = context.Process(target=update_project, args=(path, "second", other_ready, release)) + children.append(retry) + retry.start() + retry.join(10) + assert retry.exitcode == 0 + saved = SleepState.load(path) + assert set(saved.data["last_harvest"]) == {"first", "second"} + assert [row["project"] for row in saved.data["history"]] == ["first", "second"] + finally: + release.set() + for child in children: + if child.pid is not None: + if child.is_alive(): + child.terminate() + child.join(5) + + +def test_lock_releases_after_exception(tmp_path): + path = str(tmp_path / "state.json") + with pytest.raises(ValueError): + with SleepState.lock(path): + raise ValueError("synthetic cycle failure") + with SleepState.lock(path): + SleepState.load(path).save() + + +@pytest.mark.parametrize("failure", ["dump", "replace"]) +def test_interrupted_save_preserves_prior_state_and_other_tempfiles(tmp_path, monkeypatch, failure): + path = tmp_path / "state.json" + path.write_text('{"night": 4}', encoding="utf-8") + other = tmp_path / "state.json.tmp" + other.write_bytes(b"another writer's file") + state = SleepState.load(str(path)) + state.begin_night() + + def fail(*args, **kwargs): + raise OSError("synthetic interrupted save") + + monkeypatch.setattr(state_module.json if failure == "dump" else state_module.os, failure, fail) + with pytest.raises(OSError, match="interrupted save"): + state.save() + assert path.read_text(encoding="utf-8") == '{"night": 4}' + assert other.read_bytes() == b"another writer's file" + assert set(tmp_path.iterdir()) == {path, other} + + +@pytest.mark.parametrize("invalid", ["{", "[]"]) +def test_corrupt_state_is_refused_without_replacement(tmp_path, invalid): + path = tmp_path / "state.json" + path.write_text(invalid, encoding="utf-8") + with pytest.raises(ValueError): + SleepState.load(str(path)) + assert path.read_text(encoding="utf-8") == invalid + + +def test_cycle_and_handoff_share_the_transaction_lock(tmp_path, monkeypatch): + from skillopt_sleep import __main__ as cli + from skillopt_sleep import cycle + + cfg = SimpleNamespace(state_path=str(tmp_path / "state.json")) + marker = object() + + def locked(*args, **kwargs): + with pytest.raises(RuntimeError, match="busy"): + with SleepState.lock(cfg.state_path): + pytest.fail("writer started without the shared lock") + return marker + + monkeypatch.setattr(cycle, "_run_sleep_cycle", locked) + monkeypatch.setattr(cli, "_handoff_mine_and_pin_locked", locked) + assert cycle.run_sleep_cycle(cfg) is marker + assert cli._handoff_mine_and_pin(cfg, None, None, "synthetic", False) is marker From b19ccffd7a0cc2a21731520661b391470ef453bc Mon Sep 17 00:00:00 2001 From: Ryan Duguid Date: Sun, 18 Oct 2026 14:19:35 +1100 Subject: [PATCH 2/2] Read legacy locale-encoded sleep state before rewriting it as UTF-8 State written before this change used the locale codec. Fall back to it when the file is not UTF-8, and name the file in the error when neither codec decodes it. --- skillopt_sleep/state.py | 19 +++++++++++++++++-- tests/test_sleep_state_concurrency.py | 23 +++++++++++++++++++++++ 2 files changed, 40 insertions(+), 2 deletions(-) diff --git a/skillopt_sleep/state.py b/skillopt_sleep/state.py index 84345f00b..746ac2769 100644 --- a/skillopt_sleep/state.py +++ b/skillopt_sleep/state.py @@ -13,6 +13,7 @@ import copy import json +import locale import os import tempfile from contextlib import contextmanager @@ -86,10 +87,24 @@ def lock(path: str) -> Iterator[None]: @classmethod def load(cls, path: str) -> "SleepState": try: - with open(path, encoding="utf-8") as f: - data = json.load(f) + with open(path, "rb") as f: + raw = f.read() except FileNotFoundError: return cls(path, copy.deepcopy(DEFAULT_STATE)) + try: + text = raw.decode("utf-8") + except UnicodeDecodeError as utf8_error: + # Earlier writers used the locale codec (for example GBK or cp1252); + # the next save() rewrites the file as UTF-8. + codec = locale.getpreferredencoding(False) + try: + text = raw.decode(codec) + except (UnicodeDecodeError, LookupError): + raise ValueError( + f"{path} is neither UTF-8 nor {codec}; it was left unchanged. " + "Convert it to UTF-8 or move it aside to start a fresh state." + ) from utf8_error + data = json.loads(text) if not isinstance(data, dict): raise ValueError("Sleep state must contain a JSON object") merged = copy.deepcopy(DEFAULT_STATE) diff --git a/tests/test_sleep_state_concurrency.py b/tests/test_sleep_state_concurrency.py index 3dc0c08a7..20e769f4a 100644 --- a/tests/test_sleep_state_concurrency.py +++ b/tests/test_sleep_state_concurrency.py @@ -1,4 +1,5 @@ """Only one cycle may read/update/save a shared state at a time.""" +import json import multiprocessing import os from types import SimpleNamespace @@ -120,6 +121,28 @@ def test_corrupt_state_is_refused_without_replacement(tmp_path, invalid): assert path.read_text(encoding="utf-8") == invalid +@pytest.mark.parametrize("codec, note", [("gbk", "计划"), ("cp1252", "café")]) +def test_legacy_locale_state_loads_and_saves_as_utf8(tmp_path, monkeypatch, codec, note): + path = tmp_path / "state.json" + path.write_bytes(json.dumps({"night": 4, "note": note}, ensure_ascii=False).encode(codec)) + monkeypatch.setattr(state_module.locale, "getpreferredencoding", lambda do_setlocale=True: codec) + state = SleepState.load(str(path)) + assert state.data["night"] == 4 + assert state.data["note"] == note + state.save() + assert json.loads(path.read_text(encoding="utf-8"))["note"] == note + + +def test_undecodable_state_names_the_file_and_is_left_unchanged(tmp_path, monkeypatch): + path = tmp_path / "state.json" + original = b'{"note": "\xff\xfe\x81"}' + path.write_bytes(original) + monkeypatch.setattr(state_module.locale, "getpreferredencoding", lambda do_setlocale=True: "ascii") + with pytest.raises(ValueError, match="state.json is neither UTF-8 nor ascii"): + SleepState.load(str(path)) + assert path.read_bytes() == original + + def test_cycle_and_handoff_share_the_transaction_lock(tmp_path, monkeypatch): from skillopt_sleep import __main__ as cli from skillopt_sleep import cycle