task_manager_agent/task_store.py

"""
task_store.py — Task Manager Agent
=====================================
Local JSON-based task store with deduplication, status tracking,
and a simple query interface.
"""

import json
import os
import uuid
import logging
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional

logger = logging.getLogger("TaskManagerAgent.TaskStore")

TASKS_FILE = os.getenv("LOCAL_TASKS_FILE", "tasks.json")
MAX_LOCAL_TASKS = 500   # rotate after this many


class TaskStore:
    def __init__(self, filepath: str = TASKS_FILE):
        self.path = Path(filepath)
        self._tasks: list[dict] = self._load()

    # ── I/O ──────────────────────────────────────────────────────────────────

    def _load(self) -> list[dict]:
        if self.path.exists():
            try:
                with open(self.path) as f:
                    data = json.load(f)
                    return data if isinstance(data, list) else []
            except Exception as e:
                logger.warning(f"Could not load tasks file: {e}")
        return []

    def _persist(self):
        try:
            with open(self.path, "w") as f:
                json.dump(self._tasks[-MAX_LOCAL_TASKS:], f, indent=2, default=str)
        except Exception as e:
            logger.error(f"Could not persist tasks: {e}")

    # ── public API ────────────────────────────────────────────────────────────

    def save_tasks(self, tasks: list[dict]) -> int:
        """Upsert tasks by title similarity. Returns count of new tasks added."""
        existing_titles = {t.get("title", "").lower().strip() for t in self._tasks}
        added = 0
        for task in tasks:
            title_key = task.get("title", "").lower().strip()
            if title_key in existing_titles:
                logger.debug(f"Skipping duplicate task: {task.get('title')}")
                continue
            task["id"] = task.get("id") or str(uuid.uuid4())
            task["created_at"] = datetime.now(timezone.utc).isoformat()
            task["status"] = task.get("status", "todo")
            self._tasks.append(task)
            existing_titles.add(title_key)
            added += 1

        if added:
            self._persist()
        return added

    def get_tasks(
        self,
        status: Optional[str] = None,
        category: Optional[str] = None,
        priority_min: int = 0,
    ) -> list[dict]:
        result = self._tasks
        if status:
            result = [t for t in result if t.get("status") == status]
        if category:
            result = [t for t in result if t.get("category") == category]
        if priority_min:
            result = [t for t in result if (t.get("priority_score") or 0) >= priority_min]
        return sorted(result, key=lambda x: x.get("priority_score", 0), reverse=True)

    def mark_done(self, task_id: str) -> bool:
        for t in self._tasks:
            if t.get("id") == task_id:
                t["status"] = "done"
                t["completed_at"] = datetime.now(timezone.utc).isoformat()
                self._persist()
                return True
        return False

    def get_open_count(self) -> int:
        return sum(1 for t in self._tasks if t.get("status") == "todo")

    def all_as_context(self) -> list[dict]:
        """Lightweight list for LLM context / dedup checks."""
        return [{"title": t.get("title"), "status": t.get("status")} for t in self._tasks]