diff --git a/packages/core/src/agent_memory/core/ledger.py b/packages/core/src/agent_memory/core/ledger.py index f7d1c37b..fbafba3c 100644 --- a/packages/core/src/agent_memory/core/ledger.py +++ b/packages/core/src/agent_memory/core/ledger.py @@ -8,9 +8,11 @@ from __future__ import annotations import dataclasses -import pathlib import re +from .locking import store_lock +from .paths import StoreLayout + LEDGER_FILENAME = "decisions.md" VERDICT_ACCEPTED = "accepted" VERDICT_REJECTED = "rejected" @@ -41,8 +43,9 @@ def render(self) -> str: class DecisionLedger: - def __init__(self, path: pathlib.Path): - self._path = path + def __init__(self, layout: StoreLayout): + self._layout = layout + self._path = layout.dream_reports / LEDGER_FILENAME def decided(self) -> dict[str, Decision]: if not self._path.exists(): @@ -61,7 +64,14 @@ def decided(self) -> dict[str, Decision]: return found def append(self, decision: Decision) -> Decision: - self._path.parent.mkdir(parents=True, exist_ok=True) - head = self._path.read_text(encoding="utf-8") if self._path.exists() else HEADING + "\n\n" - self._path.write_text(head + decision.render() + "\n", encoding="utf-8") + with store_lock(self._layout): + self._path.parent.mkdir(parents=True, exist_ok=True) + head = ( + self._path.read_text(encoding="utf-8") + if self._path.exists() + else HEADING + "\n\n" + ) + staged = self._path.with_name(self._path.name + ".pending") + staged.write_text(head + decision.render() + "\n", encoding="utf-8") + staged.replace(self._path) return decision diff --git a/packages/core/src/agent_memory/core/manage.py b/packages/core/src/agent_memory/core/manage.py index 0f444ebd..2684cc43 100644 --- a/packages/core/src/agent_memory/core/manage.py +++ b/packages/core/src/agent_memory/core/manage.py @@ -23,7 +23,7 @@ from .clock import Clock from .database import Database from .errors import FieldError, MemoryStoreError, NotFoundError, ValidationError -from .ledger import LEDGER_FILENAME, VERDICT_ACCEPTED, VERDICT_REJECTED, Decision, DecisionLedger +from .ledger import VERDICT_ACCEPTED, VERDICT_REJECTED, Decision, DecisionLedger from .pending import Pending from .record import DATE_FIELDS, MemoryRecord from .sessions import Pointer, parse_pointer @@ -351,7 +351,7 @@ def _entry(self, name: str) -> MemoryRecord: return record def _ledger(self) -> DecisionLedger: - return DecisionLedger(self._store.layout.dream_reports / LEDGER_FILENAME) + return DecisionLedger(self._store.layout) def _usage(self) -> tuple[dict[str, int], dict[str, int], dict[str, str]]: """Lifetime counts decide what was never useful; only new reads earn weight, so one diff --git a/tests/system/test_concurrency.py b/tests/system/test_concurrency.py index a2db2e26..a2698d79 100644 --- a/tests/system/test_concurrency.py +++ b/tests/system/test_concurrency.py @@ -2,8 +2,11 @@ import multiprocessing as mp +from agent_memory.core.ledger import Decision, DecisionLedger from agent_memory.core.store import Store +DECISIONS_PER_WRITER = 6 + def _write(root: str, name: str) -> None: store = Store(root, agent=name) @@ -34,3 +37,31 @@ def test_two_processes_recording_at_once_lose_nothing(tmp_path): assert {"writer-alpha", "writer-beta"} <= names index_lines = store.layout.memory_index.read_text(encoding="utf-8") assert "writer-alpha" in index_lines and "writer-beta" in index_lines + + +def _append(root: str, name: str) -> None: + ledger = DecisionLedger(Store(root).layout) + for step in range(DECISIONS_PER_WRITER): + ledger.append( + Decision(proposal_id=f"{name}-{step}", verdict="rejected", at="2026-01-01T00:00:00Z") + ) + + +def test_two_processes_appending_decisions_lose_nothing(tmp_path): + root = tmp_path / "store" + Store(root).init() + context = mp.get_context("spawn") + writers = ("writer-alpha", "writer-beta", "writer-gamma") + workers = [ + context.Process(target=_append, args=(str(root), name)) + for name in writers + ] + for worker in workers: + worker.start() + for worker in workers: + worker.join() + assert [worker.exitcode for worker in workers] == [0, 0, 0] + + decided = DecisionLedger(Store(root).layout).decided() + expected = {f"{name}-{step}" for name in writers for step in range(DECISIONS_PER_WRITER)} + assert expected <= set(decided)