Cortex-Inara/cortex/tools/tasks.py

"""
Personal task management tools for Inara.

Tasks are stored in inara/TASKS.json — private to each agent instance.
Schema per task:
  {
    "id":          short random string,
    "title":       str,
    "description": str | None,
    "status":      "todo" | "in_progress" | "done",
    "priority":    "low" | "normal" | "high",
    "created_at":  ISO 8601,
    "updated_at":  ISO 8601
  }
"""

import json
import secrets
import asyncio
from datetime import datetime, timezone
from pathlib import Path

from google.genai import types
from persona import persona_path


def _tasks_path() -> Path:
    return persona_path() / "TASKS.json"


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _load() -> list[dict]:
    p = _tasks_path()
    if not p.exists():
        return []
    try:
        return json.loads(p.read_text())
    except Exception:
        return []


def _save(tasks: list[dict]) -> None:
    _tasks_path().write_text(json.dumps(tasks, indent=2) + "\n")


def _short_id() -> str:
    return "t_" + secrets.token_urlsafe(6)


def _format_task(t: dict) -> str:
    pri = f"[{t['priority']}]" if t.get("priority") != "normal" else ""
    desc = f"\n   {t['description']}" if t.get("description") else ""
    return f"• [{t['status']}] {t['id']} {pri} {t['title']}{desc}".strip()


# ---------------------------------------------------------------------------
# Sync implementations — called via asyncio.to_thread
# ---------------------------------------------------------------------------

def _task_list(status: str | None, priority: str | None) -> str:
    tasks = _load()
    if status:
        tasks = [t for t in tasks if t["status"] == status]
    if priority:
        tasks = [t for t in tasks if t.get("priority") == priority]
    if not tasks:
        filters = " ".join(f for f in [status, priority] if f)
        return f"No {filters} tasks." if filters else "No tasks yet."
    lines = [f"Tasks ({len(tasks)}):\n"]
    for t in tasks:
        lines.append(_format_task(t))
    return "\n".join(lines)


def _task_create(title: str, description: str | None, priority: str) -> str:
    if priority not in ("low", "normal", "high"):
        priority = "normal"
    tasks = _load()
    task = {
        "id": _short_id(),
        "title": title,
        "description": description,
        "status": "todo",
        "priority": priority,
        "created_at": _now(),
        "updated_at": _now(),
    }
    tasks.append(task)
    _save(tasks)
    return f"Created: {_format_task(task)}"


def _task_update(task_id: str, status: str | None, title: str | None,
                 description: str | None, priority: str | None) -> str:
    tasks = _load()
    for t in tasks:
        if t["id"] == task_id:
            if status and status in ("todo", "in_progress", "done"):
                t["status"] = status
            if title:
                t["title"] = title
            if description is not None:
                t["description"] = description
            if priority and priority in ("low", "normal", "high"):
                t["priority"] = priority
            t["updated_at"] = _now()
            _save(tasks)
            return f"Updated: {_format_task(t)}"
    return f"Task not found: {task_id}"


def _task_complete(task_id: str) -> str:
    return _task_update(task_id, status="done", title=None, description=None, priority=None)


# ---------------------------------------------------------------------------
# Async wrappers
# ---------------------------------------------------------------------------

async def task_list(status: str | None = None, priority: str | None = None) -> str:
    return await asyncio.to_thread(_task_list, status, priority)


async def task_create(title: str, description: str | None = None,
                      priority: str = "normal") -> str:
    return await asyncio.to_thread(_task_create, title, description, priority)


async def task_update(task_id: str, status: str | None = None, title: str | None = None,
                      description: str | None = None, priority: str | None = None) -> str:
    return await asyncio.to_thread(_task_update, task_id, status, title, description, priority)


async def task_complete(task_id: str) -> str:
    return await asyncio.to_thread(_task_complete, task_id)


DECLARATIONS = [
    types.FunctionDeclaration(
        name="task_list",
        description=(
            "List personal tasks from Inara's task list. "
            "Use this to check what's on the list, review pending work, or find a task ID. "
            "Optionally filter by status: 'todo', 'in_progress', or 'done'."
        ),
        parameters=types.Schema(
            type=types.Type.OBJECT,
            properties={
                "status": types.Schema(type=types.Type.STRING, description="Filter by status: 'todo', 'in_progress', or 'done'. Omit to list all."),
                "priority": types.Schema(type=types.Type.STRING, description="Filter by priority: 'low', 'normal', or 'high'. Omit to list all priorities."),
            },
        ),
    ),
    types.FunctionDeclaration(
        name="task_create",
        description=(
            "Add a new task to Inara's personal task list. "
            "Use this when the user asks to remember something, add a to-do, or track a follow-up."
        ),
        parameters=types.Schema(
            type=types.Type.OBJECT,
            properties={
                "title": types.Schema(type=types.Type.STRING, description="Short task title"),
                "description": types.Schema(type=types.Type.STRING, description="Optional longer description or context"),
                "priority": types.Schema(type=types.Type.STRING, description="Priority: 'low', 'normal', or 'high'. Default: normal."),
            },
            required=["title"],
        ),
    ),
    types.FunctionDeclaration(
        name="task_update",
        description=(
            "Update an existing task. Use task_list first to get the task ID. "
            "Can update status, title, description, or priority. "
            "To just mark complete, use task_complete instead."
        ),
        parameters=types.Schema(
            type=types.Type.OBJECT,
            properties={
                "task_id": types.Schema(type=types.Type.STRING, description="Task ID (e.g. t_abc123) — get from task_list"),
                "status": types.Schema(type=types.Type.STRING, description="New status: 'todo', 'in_progress', or 'done'"),
                "title": types.Schema(type=types.Type.STRING, description="Updated title"),
                "description": types.Schema(type=types.Type.STRING, description="Updated description"),
                "priority": types.Schema(type=types.Type.STRING, description="Updated priority: 'low', 'normal', or 'high'"),
            },
            required=["task_id"],
        ),
    ),
    types.FunctionDeclaration(
        name="task_complete",
        description=(
            "Mark a task as done. Use task_list first to get the task ID. "
            "Shorthand for task_update with status='done'."
        ),
        parameters=types.Schema(
            type=types.Type.OBJECT,
            properties={
                "task_id": types.Schema(type=types.Type.STRING, description="Task ID (e.g. t_abc123) — get from task_list"),
            },
            required=["task_id"],
        ),
    ),
]