code/tools/todos.py

"""Todo list tool — Claude Code-style task tracking.

Todos are persisted per-session in memory. Each todo has:
- id (string)
- content (string)
- status: pending | in_progress | completed
- priority: high | medium | low
"""

from __future__ import annotations

import threading
from typing import Any

# ─── Per-session state ──────────────────────────────────────────────────

_sessions: dict[str, list[dict[str, Any]]] = {}
_lock = threading.Lock()


def _get_session_id(session_id: str | None) -> str:
    return session_id or "default"


def todo_read(session_id: str | None = None) -> dict[str, Any]:
    """Read the current todo list for a session."""
    sid = _get_session_id(session_id)
    with _lock:
        todos = list(_sessions.get(sid, []))
    return {"success": True, "todos": todos}


def todo_write(
    todos: list[dict[str, Any]],
    session_id: str | None = None,
) -> dict[str, Any]:
    """Replace the entire todo list for a session.

    Each todo: {id, content, status, priority}
    """
    sid = _get_session_id(session_id)

    # Validate and normalize
    normalized: list[dict[str, Any]] = []
    for t in todos:
        if not isinstance(t, dict):
            continue
        normalized.append({
            "id": str(t.get("id", "")),
            "content": str(t.get("content", "")),
            "status": t.get("status", "pending") if t.get("status") in {"pending", "in_progress", "completed"} else "pending",
            "priority": t.get("priority", "medium") if t.get("priority") in {"high", "medium", "low"} else "medium",
        })

    with _lock:
        _sessions[sid] = normalized

    return {"success": True, "todos": normalized, "count": len(normalized)}


def todo_update(
    todo_id: str,
    status: str | None = None,
    content: str | None = None,
    session_id: str | None = None,
) -> dict[str, Any]:
    """Update a single todo by id."""
    sid = _get_session_id(session_id)
    with _lock:
        todos = _sessions.get(sid, [])
        for t in todos:
            if t["id"] == todo_id:
                if status in {"pending", "in_progress", "completed"}:
                    t["status"] = status
                if content is not None:
                    t["content"] = content
                return {"success": True, "todo": t}
        return {"success": False, "error": f"Todo not found: {todo_id}"}


def todo_clear(session_id: str | None = None) -> dict[str, Any]:
    """Clear all todos for a session."""
    sid = _get_session_id(session_id)
    with _lock:
        _sessions.pop(sid, None)
    return {"success": True}