diff --git a/nanobot/agent/memory.py b/nanobot/agent/memory.py index e46d6cda1..f8d93f7ed 100644 --- a/nanobot/agent/memory.py +++ b/nanobot/agent/memory.py @@ -61,6 +61,7 @@ class MemoryStore: self._cursor_file = self.memory_dir / ".cursor" self._dream_cursor_file = self.memory_dir / ".dream_cursor" self._corruption_logged = False # rate-limit non-int cursor warning + self._malformed_entry_logged = False # rate-limit bad history shape warning self._oversize_logged = False # rate-limit oversized-entry warning self._append_lock = threading.Lock() # serialize cursor allocation + append self._git = GitStore(workspace, tracked_files=[ @@ -295,8 +296,9 @@ class MemoryStore: return value def _iter_valid_entries(self) -> Iterator[tuple[dict[str, Any], int]]: - """Yield ``(entry, cursor)`` for entries with int cursors; warn once on corruption.""" + """Yield ``(entry, cursor)`` for well-formed entries; warn once on corruption.""" poisoned: Any = None + malformed_cursor: int | None = None for entry in self._read_entries(): raw = entry.get("cursor") if raw is None: @@ -305,6 +307,9 @@ class MemoryStore: if cursor is None: poisoned = raw continue + if not self._valid_history_payload(entry): + malformed_cursor = cursor + continue yield entry, cursor if poisoned is not None and not self._corruption_logged: self._corruption_logged = True @@ -313,6 +318,22 @@ class MemoryStore: "Usually caused by an external writer; further occurrences suppressed.", poisoned, ) + if malformed_cursor is not None and not self._malformed_entry_logged: + self._malformed_entry_logged = True + logger.warning( + "history.jsonl contains a malformed entry at cursor {}; dropping it. " + "Usually caused by an external writer; further occurrences suppressed.", + malformed_cursor, + ) + + @staticmethod + def _valid_history_payload(entry: dict[str, Any]) -> bool: + if not isinstance(entry.get("timestamp"), str): + return False + if not isinstance(entry.get("content"), str): + return False + session_key = entry.get("session_key") + return session_key is None or isinstance(session_key, str) def _next_cursor(self) -> int: """Read the current cursor counter and return the next value.""" diff --git a/tests/agent/test_dream.py b/tests/agent/test_dream.py index 937b7ae41..552c1ce79 100644 --- a/tests/agent/test_dream.py +++ b/tests/agent/test_dream.py @@ -87,6 +87,21 @@ class TestBuildDreamPrompt: assert "entry-21" in next_prompt assert "entry-25" in next_prompt + def test_skips_malformed_history_entries(self, store): + """Dream prompt building should tolerate externally corrupted JSONL rows.""" + store.history_file.write_text( + '{"cursor": 1, "timestamp": "2026-04-01 10:00"}\n' + '{"cursor": 2, "timestamp": "2026-04-01 10:01", "content": "usable memory"}\n', + encoding="utf-8", + ) + + result = store.build_dream_prompt() + + assert result is not None + prompt, cursor = result + assert cursor == 2 + assert "usable memory" in prompt + def test_dream_prompt_consumes_consolidator_attribute_tags(self): prompt = render_template( "agent/dream.md", diff --git a/tests/agent/test_memory_store.py b/tests/agent/test_memory_store.py index a9b5d1003..239c62a8d 100644 --- a/tests/agent/test_memory_store.py +++ b/tests/agent/test_memory_store.py @@ -171,6 +171,23 @@ class TestHistoryWithCursor: entries = store.read_unprocessed_history(since_cursor=0) assert [e["cursor"] for e in entries] == [2, 3] + def test_read_unprocessed_skips_malformed_history_payloads(self, store): + """Externally edited JSONL can keep an int cursor but miss required payload fields.""" + store.history_file.write_text( + '{"cursor": 1, "timestamp": "2026-04-01 10:00", "content": "valid"}\n' + '{"cursor": 2, "timestamp": "2026-04-01 10:01"}\n' + '{"cursor": 3, "content": "missing timestamp"}\n' + '{"cursor": 4, "timestamp": "2026-04-01 10:03", "content": 123}\n' + '{"cursor": 5, "timestamp": "2026-04-01 10:04", "content": "bad session", "session_key": 42}\n' + '{"cursor": 6, "timestamp": "2026-04-01 10:05", "content": "also valid", "session_key": "telegram:chat-1"}\n', + encoding="utf-8", + ) + + entries = store.read_unprocessed_history(since_cursor=0) + + assert [e["cursor"] for e in entries] == [1, 6] + assert [e["content"] for e in entries] == ["valid", "also valid"] + def test_next_cursor_falls_back_when_last_entry_has_no_cursor(self, store): """Regression: _next_cursor should not KeyError on entries without cursor.""" store.history_file.write_text(