"""Project-level persistent todo tools — survive across sessions.""" from __future__ import annotations import json import logging import os import threading from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable, cast logger = logging.getLogger(__name__) _LOCK = threading.RLock() _MAX = 50 _TODO_PRIORITIES = ("high", "medium", "low") _PRIORITY_RANK = {"high": 0, "medium": 1, "low": 2} def _get_workspace() -> Path | None: from butler.tools.project_todos_ops import get_active_project_workspace_safe return cast("Path | None", get_active_project_workspace_safe()) def _todos_path(workspace: Path) -> Path: return workspace / ".butler" / "todos.json" def _load(workspace: Path) -> list[dict[str, str]]: path = _todos_path(workspace) if not path.is_file(): return [] try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return [] items = data.get("items") if isinstance(data, dict) else None if not isinstance(items, list): return [] out: list[dict[str, str]] = [] for row in items: if isinstance(row, dict) and row.get("content"): pr = str(row.get("priority") or "medium").strip().lower() if pr not in _TODO_PRIORITIES: pr = "medium" out.append({ "id": str(row.get("id") or ""), "content": str(row.get("content") or ""), "status": str(row.get("status") or "pending"), "priority": pr, }) return sorted( out, key=lambda t: ( _PRIORITY_RANK.get(str(t.get("priority") or "medium"), 1), str(t.get("id") or ""), ), ) def _save(workspace: Path, items: list[dict[str, str]]) -> dict[str, Any]: record = { "project": workspace.name, "updated_at": datetime.now(timezone.utc).isoformat(), "items": items[:_MAX], } path = _todos_path(workspace) path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(".json.tmp") payload = json.dumps(record, ensure_ascii=False, indent=2) with _LOCK: try: tmp.write_text(payload, encoding="utf-8") os.replace(tmp, path) except OSError as exc: logger.warning("Project todos write failed: %s", exc) try: tmp.unlink(missing_ok=True) except OSError: pass return {"ok": False, "error": str(exc)} return {"ok": True, "count": len(items[:_MAX])} def _normalize_item(raw: Any, position: int) -> dict[str, str] | None: if isinstance(raw, str): text = raw.strip() if not text: return None return {"id": str(position), "content": text[:500], "status": "pending", "priority": "medium"} if not isinstance(raw, dict): return None content = str(raw.get("content") or raw.get("text") or "").strip() if not content: return None status = str(raw.get("status") or "pending").strip().lower() if status not in ("pending", "in_progress", "completed", "cancelled"): status = "pending" priority = str(raw.get("priority") or "medium").strip().lower() if priority not in _TODO_PRIORITIES: priority = "medium" item_id = str(raw.get("id") or position).strip() or str(position) return {"id": item_id[:32], "content": content[:500], "status": status, "priority": priority} def _tool_project_todos_list(**_: Any) -> str: ws = _get_workspace() if ws is None: return json.dumps({"ok": False, "error": "NO_ACTIVE_PROJECT"}) items = _load(ws) open_count = sum(1 for t in items if t.get("status") in ("pending", "in_progress")) return json.dumps({ "ok": True, "project": ws.name, "count": len(items), "open_count": open_count, "items": items, }, ensure_ascii=False) def _tool_project_todos_write(items: list[Any] | None = None, merge: bool = False, **_: Any) -> str: ws = _get_workspace() if ws is None: return json.dumps({"ok": False, "error": "NO_ACTIVE_PROJECT"}) if not isinstance(items, list): return json.dumps({"ok": False, "error": "items must be an array"}) if merge: existing = {str(t.get("id") or ""): dict(t) for t in _load(ws)} for raw in items: if isinstance(raw, dict): iid = str(raw.get("id") or "").strip() if iid and iid in existing: for k in ("content", "status", "priority"): v = str(raw.get(k) or "").strip() if v: existing[iid][k] = v continue item = _normalize_item(raw, len(existing) + 1) if item: existing[item["id"]] = item normalized = list(existing.values()) else: normalized = [] for i, raw in enumerate(items): if len(normalized) >= _MAX: break item = _normalize_item(raw, i + 1) if item: normalized.append(item) result = _save(ws, normalized) if not result.get("ok"): return json.dumps(result) return json.dumps({ "ok": True, "project": ws.name, "count": result["count"], "mode": "merge" if merge else "replace", "wechat": "主公可发 /项目待办 查看", }, ensure_ascii=False) def format_project_todos_for_wechat(workspace: Path, *, limit: int = 10) -> str: items = _load(workspace) if not items: return "项目待办: (空)" open_items = [t for t in items if t.get("status") in ("pending", "in_progress")] done_items = [t for t in items if t.get("status") not in ("pending", "in_progress")] lines = [f"项目待办（{len(open_items)} 进行中 / {len(done_items)} 已完成）:"] for row in open_items[:limit]: mark = "…" if row.get("status") == "in_progress" else "○" pr = row.get("priority") or "medium" lines.append(f" {mark} [{pr}] {row.get('content', '')[:80]}") if len(open_items) > limit: lines.append(f" … 另有 {len(open_items) - limit} 条") return "\n".join(lines) def register_project_todos_tools(register: Callable[..., None]) -> None: register( name="project_todos_list", description=( "List the persistent project-level todo checklist. " "Unlike session_todos, these survive across sessions." ), schema={"type": "object", "properties": {}}, handler=_tool_project_todos_list, toolset="planning", ) register( name="project_todos_write", description=( "【persistent·project JSON】Replace or merge project-level todos on disk. " "merge=true updates by id. Survives gateway restarts and /new." ), schema={ "type": "object", "properties": { "items": { "type": "array", "description": "Todo rows: {id?, content, status?, priority?}", "items": { "type": "object", "properties": { "id": {"type": "string"}, "content": {"type": "string"}, "status": {"type": "string", "enum": ["pending", "in_progress", "completed", "cancelled"]}, "priority": {"type": "string", "enum": ["high", "medium", "low"]}, }, "required": ["content"], }, }, "merge": {"type": "boolean", "description": "Merge by id instead of replace-all", "default": False}, }, "required": ["items"], }, handler=_tool_project_todos_write, toolset="planning", )