File size: 3,700 Bytes
58b74a0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 | """
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] |