Skip malformed memory entries when loading from JSONL

- Extract memory_entry_from_payload with field-level validation
  for memory_id, content, created_at, and tags types
- Return None for invalid entries instead of crashing on load

Author: johnteee

teaagent/memory.py

@@ -68,17 +68,30 @@ def _read_entries(self) -> List[MemoryEntry]:
                 payload = json.loads(line)
             except json.JSONDecodeError:
                 continue
-            entries.append(
-                MemoryEntry(
-                    memory_id=payload["memory_id"],
-                    content=payload["content"],
-                    tags=tuple(payload.get("tags", [])),
-                    created_at=payload.get("created_at", utc_now()),
-                )
-            )
+            entry = memory_entry_from_payload(payload)
+            if entry is not None:
+                entries.append(entry)
         return entries
 
 
+def memory_entry_from_payload(payload: Any) -> MemoryEntry | None:
+    if not isinstance(payload, dict):
+        return None
+    memory_id = payload.get("memory_id")
+    content = payload.get("content")
+    tags = payload.get("tags", [])
+    created_at = payload.get("created_at", utc_now())
+    if not isinstance(memory_id, str) or not memory_id:
+        return None
+    if not isinstance(content, str) or not content:
+        return None
+    if not isinstance(created_at, str) or not created_at:
+        return None
+    if not isinstance(tags, list) or not all(isinstance(tag, str) for tag in tags):
+        return None
+    return MemoryEntry(memory_id=memory_id, content=content, tags=tuple(tags), created_at=created_at)
+
+
 def normalize_tags(tags: tuple[str, ...]) -> tuple[str, ...]:
     return tuple(sorted({tag.strip().lower() for tag in tags if tag.strip()}))
 

tests/test_memory.py

@@ -5,6 +5,7 @@
 import tempfile
 import unittest
 from contextlib import redirect_stdout
+from pathlib import Path
 
 from teaagent import MemoryCatalog
 from teaagent.cli import main
@@ -22,6 +23,20 @@ def test_memory_catalog_add_list_search_show(self) -> None:
             self.assertEqual(catalog.search("graph")[0].tags, ("graph", "sqlite"))
             self.assertEqual(catalog.show(entry.memory_id).content, "GraphQLite uses a SQLite extension")
 
+    def test_memory_catalog_skips_malformed_entries(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            catalog = MemoryCatalog(tmp)
+            good = catalog.add("Keep this memory", tags=("valid",))
+            path = Path(tmp) / ".teaagent" / "memory.jsonl"
+            with path.open("a", encoding="utf-8") as handle:
+                handle.write("not json\n")
+                handle.write(json.dumps({"content": "missing id"}) + "\n")
+                handle.write(json.dumps({"memory_id": "bad-tags", "content": "x", "tags": [1]}) + "\n")
+
+            entries = catalog.list()
+
+            self.assertEqual([entry.memory_id for entry in entries], [good.memory_id])
+
     def test_cli_memory_commands(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             add_output = io.StringIO()