From 6d272ba477baac0d3bb26153ca02cc5788358653 Mon Sep 17 00:00:00 2001 From: WAXLYY Date: Sat, 11 Apr 2026 16:20:51 -0700 Subject: [PATCH] fix(tools): enforce ID uniqueness in TODO store during replace operations Deduplicate todo items by ID before writing to the store, keeping the last occurrence. Prevents ghost entries when the model sends duplicate IDs in a single write() call, which corrupts subsequent merge operations. Co-authored-by: WAXLYY --- tests/tools/test_todo_tool.py | 12 ++++++++++++ tools/todo_tool.py | 13 +++++++++++-- 2 files changed, 23 insertions(+), 2 deletions(-) diff --git a/tests/tools/test_todo_tool.py b/tests/tools/test_todo_tool.py index d4fd03baf..621507852 100644 --- a/tests/tools/test_todo_tool.py +++ b/tests/tools/test_todo_tool.py @@ -24,6 +24,18 @@ class TestWriteAndRead: items[0]["content"] = "MUTATED" assert store.read()[0]["content"] == "Task" + def test_write_deduplicates_duplicate_ids(self): + store = TodoStore() + result = store.write([ + {"id": "1", "content": "First version", "status": "pending"}, + {"id": "2", "content": "Other task", "status": "pending"}, + {"id": "1", "content": "Latest version", "status": "in_progress"}, + ]) + assert result == [ + {"id": "2", "content": "Other task", "status": "pending"}, + {"id": "1", "content": "Latest version", "status": "in_progress"}, + ] + class TestHasItems: def test_empty_store(self): diff --git a/tools/todo_tool.py b/tools/todo_tool.py index 9021fbc2d..b0d38a234 100644 --- a/tools/todo_tool.py +++ b/tools/todo_tool.py @@ -46,11 +46,11 @@ class TodoStore: """ if not merge: # Replace mode: new list entirely - self._items = [self._validate(t) for t in todos] + self._items = [self._validate(t) for t in self._dedupe_by_id(todos)] else: # Merge mode: update existing items by id, append new ones existing = {item["id"]: item for item in self._items} - for t in todos: + for t in self._dedupe_by_id(todos): item_id = str(t.get("id", "")).strip() if not item_id: continue # Can't merge without an id @@ -143,6 +143,15 @@ class TodoStore: return {"id": item_id, "content": content, "status": status} + @staticmethod + def _dedupe_by_id(todos: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + """Collapse duplicate ids, keeping the last occurrence in its position.""" + last_index: Dict[str, int] = {} + for i, item in enumerate(todos): + item_id = str(item.get("id", "")).strip() or "?" + last_index[item_id] = i + return [todos[i] for i in sorted(last_index.values())] + def todo_tool( todos: Optional[List[Dict[str, Any]]] = None,