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,