fix(todo): guard create_todo against 6-char id collisions

strix/tools/todo/tools.py

@@ -23,6 +23,8 @@
 _PRIORITY_RANK = {"critical": 0, "high": 1, "normal": 2, "low": 3}
 _STATUS_RANK = {"done": 0, "in_progress": 1, "pending": 2}
 
+_TODO_ID_GENERATION_ATTEMPTS = 1024
+
 
 def _todo_sort_key(todo: dict[str, Any]) -> tuple[int, int, str]:
     return (
@@ -109,6 +111,21 @@ def _get_agent_todos(agent_id: str) -> dict[str, dict[str, Any]]:
     return _todos_storage.setdefault(agent_id, {})
 
 
+def _generate_todo_id(agent_todos: dict[str, dict[str, Any]]) -> str | None:
+    """Return a 6-char id unused within this agent's todos, or None if none found.
+
+    Todo ids are ``str(uuid.uuid4())[:6]`` slugs (~16.7M space), so two todos
+    for the same agent can collide on the first six hex chars. Without this
+    check ``create_todo`` would silently overwrite the colliding todo. Mirrors
+    ``notes._generate_note_id`` (added in #630 for the shared 6-char scheme).
+    """
+    for _ in range(_TODO_ID_GENERATION_ATTEMPTS):
+        todo_id = str(uuid.uuid4())[:6]
+        if todo_id not in agent_todos:
+            return todo_id
+    return None
+
+
 def _normalize_priority(priority: str | None, default: str = "normal") -> str:
     candidate = (priority or default or "normal").lower()
     if candidate not in VALID_PRIORITIES:
@@ -303,9 +320,15 @@ async def create_todo(ctx: RunContextWrapper, todos: str) -> str:
 
         agent_todos = _get_agent_todos(agent_id)
         created: list[dict[str, Any]] = []
+        errors: list[dict[str, Any]] = []
         for task in tasks:
             task_priority = _normalize_priority(task.get("priority"))
-            todo_id = str(uuid.uuid4())[:6]
+            todo_id = _generate_todo_id(agent_todos)
+            if todo_id is None:
+                errors.append(
+                    {"title": task["title"], "error": "Failed to generate a unique todo ID"}
+                )
+                continue
             timestamp = datetime.now(UTC).isoformat()
             agent_todos[todo_id] = {
                 "title": task["title"],
@@ -324,18 +347,18 @@ async def create_todo(ctx: RunContextWrapper, todos: str) -> str:
             default=str,
         )
 
-    _persist()
-    return json.dumps(
-        {
-            "success": True,
-            "created": created,
-            "created_count": len(created),
-            "todos": _sorted_todos(agent_id),
-            "total_count": len(_get_agent_todos(agent_id)),
-        },
-        ensure_ascii=False,
-        default=str,
-    )
+    if created:
+        _persist()
+    response: dict[str, Any] = {
+        "success": len(errors) == 0,
+        "created": created,
+        "created_count": len(created),
+        "todos": _sorted_todos(agent_id),
+        "total_count": len(_get_agent_todos(agent_id)),
+    }
+    if errors:
+        response["errors"] = errors
+    return json.dumps(response, ensure_ascii=False, default=str)
 
 
 @function_tool(timeout=30)

tests/test_todo_id_generation.py

@@ -0,0 +1,145 @@
+"""Tests for per-agent todo id generation and collision handling."""
+
+from __future__ import annotations
+
+import json
+import uuid
+from typing import TYPE_CHECKING
+
+import pytest
+from agents.tool_context import ToolContext
+
+import strix.tools.todo.tools as todo_tools
+from strix.tools.todo.tools import _generate_todo_id
+
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+
+@pytest.fixture(autouse=True)
+def _reset_todos_storage(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
+    monkeypatch.setattr(todo_tools, "_todos_path", None)
+    with todo_tools._todos_io_lock:
+        todo_tools._todos_storage.clear()
+    yield
+    with todo_tools._todos_io_lock:
+        todo_tools._todos_storage.clear()
+
+
+class _FakeUUID:
+    """Minimal stand-in whose ``str()`` yields a controlled slug.
+
+    Todo ids are ``str(uuid.uuid4())[:6]``, so only the first six chars matter.
+    """
+
+    def __init__(self, value: str) -> None:
+        self._value = value
+
+    def __str__(self) -> str:
+        return self._value
+
+
+def _patch_uuid_sequence(monkeypatch: pytest.MonkeyPatch, slugs: list[str]) -> None:
+    """Make ``uuid.uuid4()`` yield ``str()`` values built from ``slugs`` in order."""
+    values = iter(f"{slug}-0000-0000-0000-000000000000" for slug in slugs)
+    monkeypatch.setattr(uuid, "uuid4", lambda: _FakeUUID(next(values)))
+
+
+def _seed_todo(agent_id: str, todo_id: str, title: str) -> None:
+    todo_tools._get_agent_todos(agent_id)[todo_id] = {
+        "title": title,
+        "description": None,
+        "priority": "normal",
+        "status": "pending",
+        "created_at": "2026-01-01T00:00:00+00:00",
+        "updated_at": "2026-01-01T00:00:00+00:00",
+        "completed_at": None,
+    }
+
+
+def _ctx(agent_id: str) -> ToolContext:
+    return ToolContext(
+        context={"agent_id": agent_id},
+        tool_name="create_todo",
+        tool_call_id="call-1",
+        tool_arguments="{}",
+    )
+
+
+def test_generate_id_returns_a_six_char_slug() -> None:
+    todo_id = _generate_todo_id({})
+    assert isinstance(todo_id, str)
+    assert len(todo_id) == 6
+
+
+def test_generate_id_skips_a_colliding_slug(monkeypatch: pytest.MonkeyPatch) -> None:
+    # First draw collides with an existing id; the generator must retry and
+    # return the next, unused slug instead of handing back the taken one.
+    _patch_uuid_sequence(monkeypatch, ["abcdef", "123456"])
+    assert _generate_todo_id({"abcdef": {}}) == "123456"
+
+
+def test_generate_id_returns_none_when_space_is_exhausted(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # Every draw collides — the generator gives up rather than looping forever
+    # or returning a taken id.
+    monkeypatch.setattr(
+        uuid,
+        "uuid4",
+        lambda: _FakeUUID("abcdef-0000-0000-0000-000000000000"),
+    )
+    assert _generate_todo_id({"abcdef": {}}) is None
+
+
+@pytest.mark.asyncio
+async def test_create_todo_does_not_overwrite_on_id_collision(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # Regression: a fresh todo whose generated id collides with an existing one
+    # must NOT clobber the existing todo. The generator retries to a free slug.
+    agent_id = "agent-1"
+    _seed_todo(agent_id, "abcdef", "keep me")
+    _patch_uuid_sequence(monkeypatch, ["abcdef", "bbccdd"])
+
+    raw = await todo_tools.create_todo.on_invoke_tool(
+        _ctx(agent_id), json.dumps({"todos": '[{"title": "new task"}]'})
+    )
+    result = json.loads(raw)
+
+    assert result["success"] is True
+    assert result["created"] == [{"todo_id": "bbccdd", "title": "new task", "priority": "normal"}]
+    todos = todo_tools._get_agent_todos(agent_id)
+    assert todos["abcdef"]["title"] == "keep me"
+    assert todos["bbccdd"]["title"] == "new task"
+    assert result["total_count"] == 2
+
+
+@pytest.mark.asyncio
+async def test_create_todo_reports_error_when_id_cannot_be_generated(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # If the (astronomically unlikely) id space is exhausted, the task is
+    # reported as an error rather than silently dropping or overwriting.
+    agent_id = "agent-2"
+    _seed_todo(agent_id, "abcdef", "keep me")
+    monkeypatch.setattr(
+        uuid,
+        "uuid4",
+        lambda: _FakeUUID("abcdef-0000-0000-0000-000000000000"),
+    )
+
+    raw = await todo_tools.create_todo.on_invoke_tool(
+        _ctx(agent_id), json.dumps({"todos": '[{"title": "new task"}]'})
+    )
+    result = json.loads(raw)
+
+    assert result["success"] is False
+    assert result["created"] == []
+    assert result["errors"] == [
+        {"title": "new task", "error": "Failed to generate a unique todo ID"}
+    ]
+    todos = todo_tools._get_agent_todos(agent_id)
+    assert todos["abcdef"]["title"] == "keep me"
+    assert result["total_count"] == 1