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 <WAXLYY@users.noreply.github.com>
This commit is contained in:
WAXLYY 2026-04-11 16:20:51 -07:00 committed by Teknium
parent 97b0cd51ee
commit 6d272ba477
2 changed files with 23 additions and 2 deletions

View file

@ -24,6 +24,18 @@ class TestWriteAndRead:
items[0]["content"] = "MUTATED" items[0]["content"] = "MUTATED"
assert store.read()[0]["content"] == "Task" 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: class TestHasItems:
def test_empty_store(self): def test_empty_store(self):

View file

@ -46,11 +46,11 @@ class TodoStore:
""" """
if not merge: if not merge:
# Replace mode: new list entirely # 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: else:
# Merge mode: update existing items by id, append new ones # Merge mode: update existing items by id, append new ones
existing = {item["id"]: item for item in self._items} 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() item_id = str(t.get("id", "")).strip()
if not item_id: if not item_id:
continue # Can't merge without an id continue # Can't merge without an id
@ -143,6 +143,15 @@ class TodoStore:
return {"id": item_id, "content": content, "status": status} 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( def todo_tool(
todos: Optional[List[Dict[str, Any]]] = None, todos: Optional[List[Dict[str, Any]]] = None,