diff --git a/strix/tools/todo/tools.py b/strix/tools/todo/tools.py index 07761f5f..135f3bc0 100644 --- a/strix/tools/todo/tools.py +++ b/strix/tools/todo/tools.py @@ -110,12 +110,19 @@ def _get_agent_todos(agent_id: str) -> dict[str, dict[str, Any]]: def _normalize_priority(priority: str | None, default: str = "normal") -> str: - candidate = (priority or default or "normal").lower() + candidate = str(priority or default or "normal").strip().lower() if candidate not in VALID_PRIORITIES: raise ValueError(f"Invalid priority. Must be one of: {', '.join(VALID_PRIORITIES)}") return candidate +def _coerce_priority(priority: str | None, default: str = "normal") -> str: + try: + return _normalize_priority(priority, default) + except ValueError: + return default + + def _sorted_todos(agent_id: str) -> list[dict[str, Any]]: todos_list = [ {**todo, "todo_id": todo_id} for todo_id, todo in _get_agent_todos(agent_id).items() @@ -285,11 +292,16 @@ async def create_todo(ctx: RunContextWrapper, todos: str) -> str: - ``description`` (str, optional): extra context or acceptance criteria. - ``priority`` (str, optional): one of ``"low"`` / - ``"normal"`` / ``"high"`` / ``"critical"``. Defaults to - ``"normal"``. + ``"normal"`` / ``"high"`` / ``"critical"``. Anything else, + including omitting it, falls back to ``"normal"`` rather + than failing. Example: ``[{"title": "Probe /admin", "priority": "high"}, {"title": "Check JWT alg=none"}]``. + + A title already on the list, or repeated within this call, is + skipped rather than duplicated; skipped titles come back under + ``skipped``. """ agent_id = _agent_id_from(ctx) try: @@ -302,13 +314,21 @@ async def create_todo(ctx: RunContextWrapper, todos: str) -> str: ) agent_todos = _get_agent_todos(agent_id) + seen = {todo["title"].strip().lower() for todo in agent_todos.values()} created: list[dict[str, Any]] = [] + skipped: list[dict[str, str]] = [] for task in tasks: - task_priority = _normalize_priority(task.get("priority")) + title = task["title"] + key = title.lower() + if key in seen: + skipped.append({"title": title, "reason": "duplicate title"}) + continue + seen.add(key) + task_priority = _coerce_priority(task.get("priority")) todo_id = str(uuid.uuid4())[:6] timestamp = datetime.now(UTC).isoformat() agent_todos[todo_id] = { - "title": task["title"], + "title": title, "description": task.get("description"), "priority": task_priority, "status": "pending", @@ -316,7 +336,7 @@ async def create_todo(ctx: RunContextWrapper, todos: str) -> str: "updated_at": timestamp, "completed_at": None, } - created.append({"todo_id": todo_id, "title": task["title"], "priority": task_priority}) + created.append({"todo_id": todo_id, "title": title, "priority": task_priority}) except (ValueError, TypeError) as e: return json.dumps( {"success": False, "error": f"Failed to create todo: {e}"}, @@ -330,6 +350,7 @@ async def create_todo(ctx: RunContextWrapper, todos: str) -> str: "success": True, "created": created, "created_count": len(created), + "skipped": skipped, "todos": _sorted_todos(agent_id), "total_count": len(_get_agent_todos(agent_id)), }, diff --git a/tests/test_todo.py b/tests/test_todo.py new file mode 100644 index 00000000..036bcacc --- /dev/null +++ b/tests/test_todo.py @@ -0,0 +1,105 @@ +from __future__ import annotations + +import json +from typing import Any + +import pytest +from agents.tool_context import ToolContext + +from strix.tools.todo import tools +from strix.tools.todo.tools import _coerce_priority, create_todo + + +@pytest.fixture(autouse=True) +def _isolate_store() -> Any: + tools._todos_storage.clear() + yield + tools._todos_storage.clear() + + +async def _create(todos: list[Any], agent_id: str = "root") -> dict[str, Any]: + ctx = ToolContext( + context={"agent_id": agent_id}, + tool_name="create_todo", + tool_call_id="call-1", + tool_arguments="{}", + ) + raw = await create_todo.on_invoke_tool(ctx, json.dumps({"todos": json.dumps(todos)})) + return json.loads(raw) # type: ignore[no-any-return] + + +def test_unknown_priority_falls_back_to_normal() -> None: + assert _coerce_priority("medium") == "normal" + assert _coerce_priority("urgent") == "normal" + assert _coerce_priority("high") == "high" + + +@pytest.mark.asyncio +async def test_one_bad_priority_no_longer_discards_the_batch() -> None: + result = await _create( + [ + {"title": "Recon", "priority": "medium"}, + {"title": "Probe /admin", "priority": "sky-high"}, + {"title": "Report"}, + ] + ) + + assert result["success"] is True + assert result["created_count"] == 3 + by_title = {c["title"]: c["priority"] for c in result["created"]} + assert by_title["Recon"] == "normal" + assert by_title["Probe /admin"] == "normal" + assert by_title["Report"] == "normal" + + +@pytest.mark.asyncio +async def test_duplicate_titles_within_a_batch_are_skipped() -> None: + result = await _create( + [ + {"title": "Subdomain enumeration"}, + {"title": "Content discovery"}, + {"title": "Subdomain enumeration"}, + {"title": "content discovery"}, + ] + ) + + assert result["created_count"] == 2 + assert {c["title"] for c in result["created"]} == { + "Subdomain enumeration", + "Content discovery", + } + assert len(result["skipped"]) == 2 + assert all(s["reason"] == "duplicate title" for s in result["skipped"]) + + +@pytest.mark.asyncio +async def test_a_title_already_on_the_list_is_not_created_again() -> None: + await _create([{"title": "Crawl with katana"}]) + result = await _create([{"title": "crawl with katana"}, {"title": "JS analysis"}]) + + assert [c["title"] for c in result["created"]] == ["JS analysis"] + assert [s["title"] for s in result["skipped"]] == ["crawl with katana"] + assert result["total_count"] == 2 + + +def test_coerce_never_raises() -> None: + assert _coerce_priority("nonsense") == "normal" + assert _coerce_priority(None) == "normal" + assert _coerce_priority("high") == "high" + for value in (2, ["high"], {"p": 1}, True): + assert _coerce_priority(value) == "normal" # type: ignore[arg-type] + + +@pytest.mark.asyncio +async def test_non_string_priority_does_not_fail_the_batch() -> None: + result = await _create( + [ + {"title": "Recon", "priority": 2}, + {"title": "Probe", "priority": ["high"]}, + {"title": "Report"}, + ] + ) + + assert result["success"] is True + assert result["created_count"] == 3 + assert {c["priority"] for c in result["created"]} == {"normal"}