Add LogStore local-persistence layer to crdt_sync (v0.2.0)

crdt_sync_dart: new domain-agnostic LogStore -- an in-memory Log with a
reactive changes stream and upsert/delete(sticky tombstone)/replaceAll,
behind an injectable LogPersistence port -- plus logToJson/logFromJson.
FileLogPersistence (atomic temp-file-then-rename, generalizing diet_guard's
LogStorageService) lives behind a separate crdt_sync_io.dart entrypoint so
the core barrel stays pure Dart / web-safe.

crdt-sync (Python): mirror dump_log/load_log/read_log/write_log sharing the
same on-disk JSON format so a log written on one side parses on the other.
Load/save only -- the reactive store is a Dart/Flutter convenience with no
Python equivalent by design (Python consumers are headless sync ticks).

Querying stays app-side: a Record's fields are opaque to the library, so
filtering/sorting/search is the caller's job over LogStore.values.

Both packages bumped to 0.2.0. 83 Dart tests + 71 Python tests (100% branch
coverage) green.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019KwUCPESvE5CjCtmUFz8xQ

crdt-sync/README.md

@@ -90,6 +90,26 @@ pull-all/merge/push-own pattern diet_guard's `_sync.py` already used, made
 domain-agnostic via the `encode`/`decode` callbacks so callers keep their own
 on-disk JSON shape instead of `crdt-sync` dictating one.
 
+## Local persistence
+
+For persisting a `Log` to disk, `crdt_sync` ships load/save helpers that share
+a canonical on-disk JSON shape with the Dart `crdt_sync_dart.LogStore`, so a log
+written by one language parses in the other:
+
+```python
+from crdt_sync import dump_log, load_log, read_log, write_log
+
+write_log(path, log)          # atomic temp-file-then-rename
+log = read_log(path)          # empty on a missing or corrupt file
+text = dump_log(log)          # -> canonical JSON string
+log = load_log(text)          # <- parse it back
+```
+
+These are load/save only. The Dart side additionally offers a *reactive*
+`LogStore` (a change stream to drive a live UI); that's a Flutter-only
+convenience and has no Python equivalent by design — Python consumers here are
+headless sync ticks.
+
 ## Development
 
 ```bash

crdt-sync/crdt_sync/__init__.py

@@ -6,6 +6,7 @@ from crdt_sync._github import GitHubSyncClient, GitHubSyncError, RepoNotFoundErr
 from crdt_sync._hlc import Hlc
 from crdt_sync._log import Log, merge_logs
 from crdt_sync._record import Field, Record, merge_field, merge_record
+from crdt_sync._store import dump_log, load_log, read_log, write_log
 from crdt_sync._sync import sync_log
 
 __all__ = [
@@ -16,8 +17,12 @@ __all__ = [
     "Log",
     "Record",
     "RepoNotFoundError",
+    "dump_log",
+    "load_log",
     "merge_field",
     "merge_logs",
     "merge_record",
+    "read_log",
     "sync_log",
+    "write_log",
 ]

crdt-sync/crdt_sync/_store.py

@@ -0,0 +1,70 @@
+"""Local JSON persistence for a ``Log``.
+
+Mirrors ``crdt_sync_dart``'s ``logToJson``/``FileLogPersistence`` serialization
+so a log written by the Dart side parses here and vice versa. Deliberately
+*non-reactive*: the change-stream/reactive store is a Dart/Flutter-only
+convenience (it exists to drive a live UI). On the Python side, where consumers
+are headless sync ticks, this is plain load and (atomic) save.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from typing import TYPE_CHECKING
+
+from crdt_sync._record import Record
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    from crdt_sync._log import Log
+
+
+def dump_log(log: Log) -> str:
+    """Serialize ``log`` to canonical JSON text (id -> record dict)."""
+    return json.dumps({rid: record.to_dict() for rid, record in log.items()})
+
+
+def load_log(text: str) -> Log:
+    """Parse text produced by :func:`dump_log` back into a ``Log``.
+
+    Raises ``ValueError`` (bad JSON), ``KeyError`` or ``TypeError`` (valid
+    JSON, wrong shape) on a corrupt payload -- callers reading from disk go
+    through :func:`read_log`, which turns those into an empty log.
+    """
+    raw = json.loads(text)
+    if not isinstance(raw, dict):
+        msg = f"expected a JSON object mapping id -> record, got {type(raw).__name__}"
+        raise TypeError(msg)
+    return {rid: Record.from_dict(value) for rid, value in raw.items()}
+
+
+def read_log(path: Path) -> Log:
+    """Read a ``Log`` from ``path``; empty on a missing or unparsable file.
+
+    A truncated or corrupt file can never raise here: it is treated as "no log
+    yet", matching the Dart store's defensive load so a half-written file can't
+    brick a consumer.
+    """
+    try:
+        text = path.read_text(encoding="utf-8")
+    except FileNotFoundError:
+        return {}
+    try:
+        return load_log(text)
+    except (ValueError, KeyError, TypeError):
+        return {}
+
+
+def write_log(path: Path, log: Log) -> None:
+    """Write ``log`` to ``path`` atomically (temp file then rename).
+
+    Writes to a per-process temp file and ``replace``s it over the real path,
+    so a concurrent reader never observes a half-written file. Mirrors the
+    Dart ``FileLogPersistence`` write scheme.
+    """
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_name(f"{path.name}.{os.getpid()}.tmp")
+    tmp.write_text(dump_log(log), encoding="utf-8")
+    tmp.replace(path)

crdt-sync/crdt_sync/tests/test_store.py

@@ -0,0 +1,83 @@
+"""Tests for local Log persistence (dump/load and atomic file read/write)."""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING
+
+import pytest
+
+from crdt_sync import (
+    Record,
+    dump_log,
+    load_log,
+    read_log,
+    write_log,
+)
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+    from pathlib import Path
+
+    from crdt_sync import Hlc, Log
+
+
+def _sample_log(make_hlc: Callable[..., Hlc]) -> Log:
+    return {
+        "a": Record(id="a", fields={"text": ("alpha", make_hlc(100))}),
+        "b": Record(id="b", fields={}, deleted=True, deleted_hlc=make_hlc(200)),
+    }
+
+
+class TestDumpLoad:
+    def test_round_trips_a_log(self, make_hlc: Callable[..., Hlc]) -> None:
+        log = _sample_log(make_hlc)
+        assert load_log(dump_log(log)) == log
+
+    def test_load_raises_on_non_json(self) -> None:
+        # json.loads raises JSONDecodeError (a ValueError subclass).
+        with pytest.raises(json.JSONDecodeError):
+            load_log("{not json")
+
+    def test_load_raises_typeerror_on_wrong_shape(self) -> None:
+        # Valid JSON, but a list where an id->record map is expected.
+        with pytest.raises(TypeError):
+            load_log("[]")
+
+
+class TestReadWrite:
+    def test_write_then_read_round_trips(
+        self, tmp_path: Path, make_hlc: Callable[..., Hlc]
+    ) -> None:
+        log = _sample_log(make_hlc)
+        path = tmp_path / "nested" / "log.json"
+        write_log(path, log)
+        assert read_log(path) == log
+
+    def test_write_creates_parent_directories(
+        self, tmp_path: Path, make_hlc: Callable[..., Hlc]
+    ) -> None:
+        path = tmp_path / "deep" / "nested" / "log.json"
+        write_log(path, _sample_log(make_hlc))
+        assert path.exists()
+
+    def test_write_overwrites_atomically(
+        self, tmp_path: Path, make_hlc: Callable[..., Hlc]
+    ) -> None:
+        path = tmp_path / "log.json"
+        write_log(path, {"a": Record(id="a", fields={})})
+        write_log(path, _sample_log(make_hlc))
+        assert set(read_log(path)) == {"a", "b"}
+
+    def test_read_of_missing_file_is_empty(self, tmp_path: Path) -> None:
+        assert read_log(tmp_path / "nope.json") == {}
+
+    def test_read_of_corrupt_file_is_empty(self, tmp_path: Path) -> None:
+        path = tmp_path / "log.json"
+        path.write_text("{not json", encoding="utf-8")
+        assert read_log(path) == {}
+
+    def test_read_of_wrong_shape_file_is_empty(self, tmp_path: Path) -> None:
+        path = tmp_path / "log.json"
+        path.write_text("[]", encoding="utf-8")
+        assert read_log(path) == {}

crdt-sync/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "crdt-sync"
-version = "0.1.0"
+version = "0.2.0"
 description = "Shared CRDT merge scheme + GitHub-Contents-API sync transport for cross-device apps"
 requires-python = ">=3.10"
 dependencies = ["requests"]

crdt_sync_dart/README.md

@@ -66,6 +66,42 @@ final merged = await syncLog(
 `<pathPrefix>/<other-device-id>/...`, merges each into the local log with
 `mergeLogs`, then pushes this device's own merged result back up.
 
+## Local persistence (`LogStore`)
+
+The merge scheme above is in-memory only. For an app that needs to keep its
+`Log` on disk and drive a live UI off it, this package also ships a
+domain-agnostic `LogStore`:
+
+```dart
+import 'package:crdt_sync/crdt_sync.dart';
+import 'package:crdt_sync/crdt_sync_io.dart'; // FileLogPersistence (dart:io)
+
+final store = LogStore(
+  persistence: FileLogPersistence(File('$docsDir/notes.json')),
+  nodeId: 'phone',
+);
+await store.load();
+
+// Reactive read: re-derive a domain view whenever anything changes. Filtering,
+// sorting and search live here, in the app -- the store is field-agnostic.
+final Stream<List<Record>> visible = store.changes.map(
+  (_) => store.values.where((r) => !r.deleted).toList(),
+);
+
+await store.upsert(Record(id: 'n1', fields: {'text': ('hi', store.nextHlc())}));
+await store.delete('n1'); // sticky tombstone, not a hard remove
+
+// After a sync tick, swap in the merged result:
+await store.replaceAll(await syncLog(/* ... */, localLog: store.snapshot()));
+```
+
+`LogStore` is pure Dart (import it from the main barrel); the filesystem-backed
+`FileLogPersistence` lives behind the separate `crdt_sync_io.dart` entrypoint so
+the core stays web-safe. Querying is deliberately **not** in the library — a
+`Record`'s fields are an opaque map, so the app filters `store.values` itself.
+The Python side mirrors the on-disk format via `crdt_sync.dump_log`/`read_log`
+(load/save only; no reactive stream — that's a Dart/Flutter convenience).
+
 ## Development
 
 ```bash

crdt_sync_dart/lib/crdt_sync.dart

@@ -5,4 +5,5 @@ export 'src/github_client.dart';
 export 'src/hlc.dart';
 export 'src/log.dart';
 export 'src/record.dart';
+export 'src/store.dart';
 export 'src/sync.dart';

crdt_sync_dart/lib/crdt_sync_io.dart

@@ -0,0 +1,8 @@
+/// Filesystem-backed persistence for `LogStore`.
+///
+/// Import this instead of (in addition to) `package:crdt_sync/crdt_sync.dart`
+/// on platforms with a filesystem (mobile, desktop, server). It pulls in
+/// `dart:io`, so it is unavailable on web; the main barrel stays pure Dart.
+library;
+
+export 'src/io_store.dart';

crdt_sync_dart/lib/src/io_store.dart

@@ -0,0 +1,40 @@
+import 'dart:io';
+
+import 'store.dart';
+
+/// A [LogPersistence] backed by a single JSON file.
+///
+/// Writes atomically -- to a per-process temp file, then `rename`d over the
+/// real path -- so a concurrent reader (e.g. a background sync isolate writing
+/// while the foreground app also writes) never observes a half-written file.
+/// The pid in the temp name keeps two writers from colliding. This generalizes
+/// diet_guard's hand-rolled `LogStorageService` write scheme.
+///
+/// Lives behind the `package:crdt_sync/crdt_sync_io.dart` entrypoint, not the
+/// main barrel, because it imports `dart:io` and so is unavailable on web --
+/// the core [LogStore] stays pure Dart.
+class FileLogPersistence implements LogPersistence {
+  /// Persists to [file]. The app chooses the path (e.g. via `path_provider`);
+  /// the library takes no dependency on where app documents live.
+  FileLogPersistence(this._file);
+
+  final File _file;
+
+  @override
+  Future<String?> read() async {
+    if (!_file.existsSync()) return null;
+    try {
+      return await _file.readAsString();
+    } on FileSystemException {
+      return null;
+    }
+  }
+
+  @override
+  Future<void> write(String text) async {
+    await _file.parent.create(recursive: true);
+    final tmp = File('${_file.path}.$pid.tmp');
+    await tmp.writeAsString(text);
+    await tmp.rename(_file.path);
+  }
+}

crdt_sync_dart/lib/src/store.dart

@@ -0,0 +1,183 @@
+import 'dart:async';
+import 'dart:convert';
+
+import 'hlc.dart';
+import 'log.dart';
+import 'record.dart';
+
+/// Serializes a [Log] to canonical JSON text: a top-level object mapping each
+/// record id to its [Record.toJson]. Mirrors `crdt_sync._store.dump_log` on
+/// the Python side, so a log written by one language parses in the other.
+String logToJson(Log log) => jsonEncode({
+  for (final entry in log.entries) entry.key: entry.value.toJson(),
+});
+
+/// Parses text produced by [logToJson] back into a [Log].
+///
+/// Throws [FormatException] (bad JSON) or [TypeError] (valid JSON, wrong
+/// shape) on a corrupt payload -- callers persisting to disk should treat
+/// both as "start empty", see [LogPersistence]-backed [LogStore.load].
+Log logFromJson(String text) {
+  final raw = jsonDecode(text) as Map<String, dynamic>;
+  return raw.map(
+    (id, value) =>
+        MapEntry(id, Record.fromJson((value as Map).cast<String, dynamic>())),
+  );
+}
+
+/// A durable, string-in/string-out persistence port for a [LogStore].
+///
+/// Kept deliberately free of `dart:io` so [LogStore] stays pure Dart (usable
+/// on web and trivially faked in tests). A filesystem-backed implementation
+/// lives behind the separate `package:crdt_sync/crdt_sync_io.dart` entrypoint;
+/// tests inject an in-memory one.
+abstract class LogPersistence {
+  /// Returns the stored text, or `null` if nothing has been persisted yet.
+  Future<String?> read();
+
+  /// Overwrites the stored text with [text].
+  Future<void> write(String text);
+}
+
+/// An in-memory [Log] with reactive change notifications, backed by a
+/// [LogPersistence] port.
+///
+/// This is the local-persistence half that the merge scheme itself
+/// (`Hlc`/`Record`/`Log`) deliberately omits. It is domain-agnostic:
+/// filtering, sorting and searching over [values] is the caller's job -- the
+/// store knows nothing about what fields a record carries. A consumer's typical
+/// reactive read is `store.changes.map((_) => store.values.where(pred))`.
+///
+/// The store owns this node's logical clock: every mutation it performs
+/// (a [delete] tombstone) is stamped with a monotonic [Hlc] for [nodeId], and
+/// callers building their own records should take field clocks from [nextHlc]
+/// so the whole device shares one monotonic sequence.
+class LogStore {
+  /// Creates a store for [nodeId] persisting through [persistence].
+  ///
+  /// Call [load] once before reading [values] to hydrate from storage.
+  // Dart forbids private named params, so these can't be initializing
+  // formals; assign them explicitly.
+  // ignore_for_file: prefer_initializing_formals
+  LogStore({required LogPersistence persistence, required String nodeId})
+    : _persistence = persistence,
+      _nodeId = nodeId;
+
+  final LogPersistence _persistence;
+  final String _nodeId;
+  final Log _log = {};
+  final StreamController<void> _changes = StreamController<void>.broadcast();
+  Hlc? _lastHlc;
+
+  /// Fires (with no payload) after every mutation that reaches storage.
+  ///
+  /// A UI listens here and re-derives its view from [values]; the event
+  /// carries nothing because the store is not the place that knows how to
+  /// shape a domain view.
+  Stream<void> get changes => _changes.stream;
+
+  /// The current records, tombstones included. Callers filter as needed.
+  Iterable<Record> get values => _log.values;
+
+  /// Returns the record for [id], or `null` if absent.
+  Record? get(String id) => _log[id];
+
+  /// An unmodifiable snapshot of the whole log, e.g. to hand to `syncLog`.
+  Log snapshot() => Map.unmodifiable(_log);
+
+  /// Returns the next monotonic [Hlc] for this node, advancing the clock.
+  ///
+  /// Callers stamp their own record fields with this so a device's writes and
+  /// the store's own delete tombstones share one strictly increasing sequence.
+  Hlc nextHlc() {
+    final next = Hlc.newTick(_nodeId, previous: _lastHlc);
+    _lastHlc = next;
+    return next;
+  }
+
+  /// Hydrates the in-memory log from storage, returning a snapshot.
+  ///
+  /// A missing or corrupt payload is treated as an empty log (never an error),
+  /// so a truncated write can't brick the app -- it just loses the unwritten
+  /// tail, exactly as the file-storage mirror in diet_guard does.
+  Future<Log> load() async {
+    final text = await _persistence.read();
+    _log.clear();
+    if (text != null) {
+      try {
+        _log.addAll(logFromJson(text));
+      } on FormatException {
+        _log.clear();
+      } on TypeError {
+        _log.clear();
+      }
+    }
+    _trackLatestHlc();
+    return snapshot();
+  }
+
+  /// Inserts or replaces [record] and persists.
+  ///
+  /// The caller owns the record's field clocks (take them from [nextHlc]);
+  /// the store does not re-stamp them, so an unchanged re-upsert is a no-op
+  /// under a later merge.
+  Future<void> upsert(Record record) async {
+    _log[record.id] = record;
+    _trackHlcsOf(record);
+    await _persist();
+  }
+
+  /// Tombstones the record with [id] in place (sticky CRDT delete) and
+  /// persists. No-op if [id] is absent or already deleted.
+  ///
+  /// Tombstoning rather than removing is what stops a delete from being
+  /// silently undone when an older, still-present copy is pulled from another
+  /// device on the next sync.
+  Future<void> delete(String id) async {
+    final existing = _log[id];
+    if (existing == null || existing.deleted) return;
+    _log[id] = Record(
+      id: existing.id,
+      fields: existing.fields,
+      deleted: true,
+      deletedHlc: nextHlc(),
+    );
+    await _persist();
+  }
+
+  /// Replaces the entire log (e.g. with a post-merge result from `syncLog`)
+  /// and persists.
+  Future<void> replaceAll(Log merged) async {
+    _log
+      ..clear()
+      ..addAll(merged);
+    _trackLatestHlc();
+    await _persist();
+  }
+
+  /// Releases the [changes] stream. The store is unusable afterwards.
+  Future<void> close() => _changes.close();
+
+  Future<void> _persist() async {
+    await _persistence.write(logToJson(_log));
+    if (!_changes.isClosed) _changes.add(null);
+  }
+
+  void _trackLatestHlc() {
+    for (final record in _log.values) {
+      _trackHlcsOf(record);
+    }
+  }
+
+  void _trackHlcsOf(Record record) {
+    for (final field in record.fields.values) {
+      _bumpLastHlc(field.$2);
+    }
+    final deletedHlc = record.deletedHlc;
+    if (deletedHlc != null) _bumpLastHlc(deletedHlc);
+  }
+
+  void _bumpLastHlc(Hlc candidate) {
+    if (_lastHlc == null || candidate > _lastHlc!) _lastHlc = candidate;
+  }
+}

crdt_sync_dart/pubspec.yaml

@@ -2,7 +2,7 @@ name: crdt_sync
 description: >-
   Shared CRDT merge scheme + GitHub-Contents-API sync transport for
   cross-device apps. Pure Dart, no Flutter dependency.
-version: 0.1.0
+version: 0.2.0
 publish_to: 'none'
 
 environment:

crdt_sync_dart/test/store_test.dart

@@ -0,0 +1,236 @@
+import 'dart:io';
+
+import 'package:crdt_sync/crdt_sync.dart';
+import 'package:crdt_sync/crdt_sync_io.dart';
+import 'package:test/test.dart';
+
+Hlc _make(int wallTimeMs, {int counter = 0, String nodeId = 'node-a'}) =>
+    Hlc(wallTimeMs: wallTimeMs, counter: counter, nodeId: nodeId);
+
+Record _rec(String id, String text, Hlc hlc) =>
+    Record(id: id, fields: {'text': (text, hlc)});
+
+/// An in-memory [LogPersistence] so the core store is testable without disk.
+class _MemoryPersistence implements LogPersistence {
+  String? text;
+  int writes = 0;
+
+  @override
+  Future<String?> read() async => text;
+
+  @override
+  Future<void> write(String value) async {
+    text = value;
+    writes++;
+  }
+}
+
+void main() {
+  group('logToJson / logFromJson', () {
+    test('round-trips a log', () {
+      final log = <String, Record>{
+        'a': _rec('a', 'alpha', _make(100)),
+        'b': Record(id: 'b', fields: {}, deleted: true, deletedHlc: _make(200)),
+      };
+      expect(logFromJson(logToJson(log)), equals(log));
+    });
+
+    test('throws FormatException on non-JSON text', () {
+      expect(() => logFromJson('{not json'), throwsFormatException);
+    });
+
+    test('throws TypeError on valid JSON of the wrong shape', () {
+      expect(() => logFromJson('[]'), throwsA(isA<TypeError>()));
+    });
+  });
+
+  group('LogStore.load', () {
+    test('is empty when nothing was ever persisted', () async {
+      final store = LogStore(persistence: _MemoryPersistence(), nodeId: 'n');
+      expect(await store.load(), isEmpty);
+      expect(store.values, isEmpty);
+    });
+
+    test('hydrates from a valid stored payload', () async {
+      final persistence = _MemoryPersistence()
+        ..text = logToJson({'a': _rec('a', 'alpha', _make(100))});
+      final store = LogStore(persistence: persistence, nodeId: 'n');
+      final log = await store.load();
+      expect(log.keys, ['a']);
+      expect(store.get('a')!.fields['text']!.$1, 'alpha');
+    });
+
+    test('treats a syntactically corrupt payload as empty', () async {
+      final persistence = _MemoryPersistence()..text = '{not json';
+      final store = LogStore(persistence: persistence, nodeId: 'n');
+      expect(await store.load(), isEmpty);
+    });
+
+    test('treats a wrong-shape payload as empty', () async {
+      final persistence = _MemoryPersistence()..text = '[]';
+      final store = LogStore(persistence: persistence, nodeId: 'n');
+      expect(await store.load(), isEmpty);
+    });
+  });
+
+  group('LogStore mutations', () {
+    test('upsert stores the record and persists', () async {
+      final persistence = _MemoryPersistence();
+      final store = LogStore(persistence: persistence, nodeId: 'n');
+      await store.load();
+      final record = _rec('a', 'alpha', store.nextHlc());
+      await store.upsert(record);
+      expect(store.get('a'), record);
+      expect(persistence.writes, 1);
+      expect(logFromJson(persistence.text!)['a'], record);
+    });
+
+    test('delete tombstones in place and persists', () async {
+      final store = LogStore(persistence: _MemoryPersistence(), nodeId: 'n');
+      await store.load();
+      await store.upsert(_rec('a', 'alpha', store.nextHlc()));
+      await store.delete('a');
+      final tombstone = store.get('a')!;
+      expect(tombstone.deleted, isTrue);
+      expect(tombstone.deletedHlc, isNotNull);
+      expect(tombstone.fields['text']!.$1, 'alpha'); // fields retained
+    });
+
+    test('delete is a no-op for an absent id', () async {
+      final persistence = _MemoryPersistence();
+      final store = LogStore(persistence: persistence, nodeId: 'n');
+      await store.load();
+      await store.delete('missing');
+      expect(persistence.writes, 0);
+    });
+
+    test('delete is a no-op for an already-tombstoned record', () async {
+      final persistence = _MemoryPersistence();
+      final store = LogStore(persistence: persistence, nodeId: 'n');
+      await store.load();
+      await store.upsert(_rec('a', 'alpha', store.nextHlc()));
+      await store.delete('a');
+      final writesAfterFirstDelete = persistence.writes;
+      await store.delete('a');
+      expect(persistence.writes, writesAfterFirstDelete);
+    });
+
+    test('replaceAll swaps the whole log and persists', () async {
+      final persistence = _MemoryPersistence();
+      final store = LogStore(persistence: persistence, nodeId: 'n');
+      await store.load();
+      await store.upsert(_rec('a', 'alpha', store.nextHlc()));
+      final merged = <String, Record>{'b': _rec('b', 'beta', _make(500))};
+      await store.replaceAll(merged);
+      expect(store.values.map((r) => r.id), ['b']);
+      expect(logFromJson(persistence.text!), merged);
+    });
+  });
+
+  group('LogStore clock & snapshot', () {
+    test('nextHlc is strictly monotonic', () async {
+      final store = LogStore(persistence: _MemoryPersistence(), nodeId: 'n');
+      final first = store.nextHlc();
+      final second = store.nextHlc();
+      expect(second > first, isTrue);
+    });
+
+    test('nextHlc stays ahead of hydrated field clocks', () async {
+      final far = _make(1 << 40); // a clock far in the future
+      final persistence = _MemoryPersistence()
+        ..text = logToJson({'a': _rec('a', 'alpha', far)});
+      final store = LogStore(persistence: persistence, nodeId: 'n');
+      await store.load();
+      expect(store.nextHlc() > far, isTrue);
+    });
+
+    test('snapshot is unmodifiable', () async {
+      final store = LogStore(persistence: _MemoryPersistence(), nodeId: 'n');
+      await store.load();
+      expect(
+        () => store.snapshot()['x'] = _rec('x', 'y', _make(1)),
+        throwsUnsupportedError,
+      );
+    });
+
+    test('get returns null for an absent id', () async {
+      final store = LogStore(persistence: _MemoryPersistence(), nodeId: 'n');
+      await store.load();
+      expect(store.get('nope'), isNull);
+    });
+  });
+
+  group('LogStore.changes', () {
+    test('fires once per mutation', () async {
+      final store = LogStore(persistence: _MemoryPersistence(), nodeId: 'n');
+      await store.load();
+      final events = <void>[];
+      final sub = store.changes.listen(events.add);
+      await store.upsert(_rec('a', 'alpha', store.nextHlc()));
+      await store.delete('a');
+      await Future<void>.delayed(Duration.zero);
+      expect(events.length, 2);
+      await sub.cancel();
+    });
+
+    test('close ends the stream and later writes do not throw', () async {
+      final persistence = _MemoryPersistence();
+      final store = LogStore(persistence: persistence, nodeId: 'n');
+      await store.load();
+      await store.close();
+      // isClosed guard: the write still lands, the (closed) stream is skipped.
+      await store.upsert(_rec('a', 'alpha', _make(1)));
+      expect(persistence.writes, 1);
+    });
+  });
+
+  group('FileLogPersistence', () {
+    late Directory dir;
+
+    setUp(() async {
+      dir = await Directory.systemTemp.createTemp('crdt_sync_store_test');
+    });
+
+    tearDown(() async {
+      if (dir.existsSync()) await dir.delete(recursive: true);
+    });
+
+    test('read returns null when the file does not exist', () async {
+      final persistence = FileLogPersistence(File('${dir.path}/nope.json'));
+      expect(await persistence.read(), isNull);
+    });
+
+    test('write then read round-trips and creates parent dirs', () async {
+      final path = '${dir.path}/nested/deep/log.json';
+      final persistence = FileLogPersistence(File(path));
+      await persistence.write('hello');
+      expect(await persistence.read(), 'hello');
+      expect(File(path).existsSync(), isTrue);
+    });
+
+    test('write overwrites atomically', () async {
+      final file = File('${dir.path}/log.json');
+      final persistence = FileLogPersistence(file);
+      await persistence.write('first');
+      await persistence.write('second');
+      expect(await persistence.read(), 'second');
+    });
+
+    test('backs a LogStore end to end', () async {
+      final file = File('${dir.path}/store.json');
+      final store = LogStore(
+        persistence: FileLogPersistence(file),
+        nodeId: 'device-1',
+      );
+      await store.load();
+      await store.upsert(_rec('a', 'alpha', store.nextHlc()));
+
+      final reopened = LogStore(
+        persistence: FileLogPersistence(file),
+        nodeId: 'device-1',
+      );
+      final log = await reopened.load();
+      expect(log['a']!.fields['text']!.$1, 'alpha');
+    });
+  });
+}