diff --git a/cortex/tools/__init__.py b/cortex/tools/__init__.py index 593f797..e8a4840 100644 --- a/cortex/tools/__init__.py +++ b/cortex/tools/__init__.py @@ -21,6 +21,8 @@ from tools.ae_knowledge import journal_entry_create as _ae_journal_entry_create from tools.ae_tasks import task_list as _ae_task_list from tools.files import file_read as _file_read from tools.system import claude_allow_dir as _claude_allow_dir +from tools.tasks import task_list as _task_list, task_create as _task_create +from tools.tasks import task_update as _task_update, task_complete as _task_complete # --------------------------------------------------------------------------- @@ -173,6 +175,10 @@ _CALLABLES: dict[str, callable] = { "ae_task_list": _ae_task_list, "file_read": _file_read, "claude_allow_dir": _claude_allow_dir, + "task_list": _task_list, + "task_create": _task_create, + "task_update": _task_update, + "task_complete": _task_complete, } _claude_allow_dir_declaration = types.FunctionDeclaration( @@ -202,6 +208,103 @@ _claude_allow_dir_declaration = types.FunctionDeclaration( ), ) +_task_list_declaration = 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.", + ), + }, + ), +) + +_task_create_declaration = 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"], + ), +) + +_task_update_declaration = 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"], + ), +) + +_task_complete_declaration = 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"], + ), +) + # Gemini Tool object — pass this to GenerateContentConfig TOOL_DECLARATIONS = [ types.Tool(function_declarations=[ @@ -211,6 +314,10 @@ TOOL_DECLARATIONS = [ _ae_task_list_declaration, _file_read_declaration, _claude_allow_dir_declaration, + _task_list_declaration, + _task_create_declaration, + _task_update_declaration, + _task_complete_declaration, ]) ] diff --git a/cortex/tools/tasks.py b/cortex/tools/tasks.py new file mode 100644 index 0000000..c62e004 --- /dev/null +++ b/cortex/tools/tasks.py @@ -0,0 +1,135 @@ +""" +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 config import settings + + +def _tasks_path() -> Path: + return settings.inara_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) -> str: + tasks = _load() + if status: + tasks = [t for t in tasks if t["status"] == status] + if not tasks: + label = f"No {status} tasks." if status else "No tasks yet." + return label + 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) -> str: + return await asyncio.to_thread(_task_list, status) + + +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)