fix(memory): preserve unprocessed dream history
This commit is contained in:
+22
-2
@@ -433,13 +433,33 @@ class MemoryStore:
|
|||||||
]
|
]
|
||||||
|
|
||||||
def compact_history(self) -> None:
|
def compact_history(self) -> None:
|
||||||
"""Drop oldest entries if the file exceeds *max_history_entries*."""
|
"""Drop oldest processed entries without discarding pending Dream input."""
|
||||||
if self.max_history_entries <= 0:
|
if self.max_history_entries <= 0:
|
||||||
return
|
return
|
||||||
entries = self._read_entries()
|
entries = self._read_entries()
|
||||||
if len(entries) <= self.max_history_entries:
|
if len(entries) <= self.max_history_entries:
|
||||||
return
|
return
|
||||||
kept = entries[-self.max_history_entries:]
|
last_dream_cursor = self.get_last_dream_cursor()
|
||||||
|
first_unprocessed = next(
|
||||||
|
(
|
||||||
|
index
|
||||||
|
for index, entry in enumerate(entries)
|
||||||
|
if (
|
||||||
|
(cursor := self._valid_cursor(entry.get("cursor"))) is not None
|
||||||
|
and cursor > last_dream_cursor
|
||||||
|
)
|
||||||
|
),
|
||||||
|
len(entries),
|
||||||
|
)
|
||||||
|
keep_from = min(len(entries) - self.max_history_entries, first_unprocessed)
|
||||||
|
kept = entries[keep_from:]
|
||||||
|
if len(kept) > self.max_history_entries:
|
||||||
|
logger.warning(
|
||||||
|
"History compaction retained {} unprocessed entries beyond the configured "
|
||||||
|
"limit of {}",
|
||||||
|
len(kept),
|
||||||
|
self.max_history_entries,
|
||||||
|
)
|
||||||
self._write_entries(kept)
|
self._write_entries(kept)
|
||||||
|
|
||||||
# -- JSONL helpers -------------------------------------------------------
|
# -- JSONL helpers -------------------------------------------------------
|
||||||
|
|||||||
@@ -235,11 +235,23 @@ class TestHistoryWithCursor:
|
|||||||
store.append_history("event 3")
|
store.append_history("event 3")
|
||||||
store.append_history("event 4")
|
store.append_history("event 4")
|
||||||
store.append_history("event 5")
|
store.append_history("event 5")
|
||||||
|
store.set_last_dream_cursor(5)
|
||||||
store.compact_history()
|
store.compact_history()
|
||||||
entries = store.read_unprocessed_history(since_cursor=0)
|
entries = store.read_unprocessed_history(since_cursor=0)
|
||||||
assert len(entries) == 2
|
assert len(entries) == 2
|
||||||
assert entries[0]["cursor"] in {4, 5}
|
assert entries[0]["cursor"] in {4, 5}
|
||||||
|
|
||||||
|
def test_compact_history_preserves_entries_after_dream_cursor(self, tmp_path):
|
||||||
|
store = MemoryStore(tmp_path, max_history_entries=50)
|
||||||
|
for index in range(1, 101):
|
||||||
|
store.append_history(f"event {index}")
|
||||||
|
store.set_last_dream_cursor(20)
|
||||||
|
|
||||||
|
store.compact_history()
|
||||||
|
|
||||||
|
entries = store.read_unprocessed_history(since_cursor=0)
|
||||||
|
assert [entry["cursor"] for entry in entries] == list(range(21, 101))
|
||||||
|
|
||||||
def test_write_entries_uses_atomic_write(self, tmp_path):
|
def test_write_entries_uses_atomic_write(self, tmp_path):
|
||||||
"""_write_entries uses temp file + os.replace for atomicity."""
|
"""_write_entries uses temp file + os.replace for atomicity."""
|
||||||
store = MemoryStore(tmp_path)
|
store = MemoryStore(tmp_path)
|
||||||
|
|||||||
Reference in New Issue
Block a user