diff --git a/strix/tools/todo/tools.py b/strix/tools/todo/tools.py index 07761f5f8..9a4cc13c0 100644 --- a/strix/tools/todo/tools.py +++ b/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) diff --git a/tests/test_todo_id_generation.py b/tests/test_todo_id_generation.py new file mode 100644 index 000000000..e174bbc23 --- /dev/null +++ b/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