""" Reminders tools. Reminders are stored in persona/REMINDERS.md and automatically surfaced in the system prompt at Tier 2+. Use these tools to add, list, and clear pending reminders. Operations: reminders_add — append a new reminder entry reminders_list — return all current reminders (or a message if empty) reminders_clear — erase all reminders (moved here from cron.py for consistency; cron.py still calls the same underlying file) """ import asyncio from datetime import datetime, timezone from pathlib import Path from persona import persona_path def _reminders_path() -> Path: return persona_path() / "REMINDERS.md" def _now_label() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") def _parse_sections(text: str) -> list[tuple[str, str]]: """Split REMINDERS.md into (heading, body) tuples, one per ## section.""" sections: list[tuple[str, str]] = [] heading: str | None = None body_lines: list[str] = [] for line in text.splitlines(): if line.startswith("## "): if heading is not None: sections.append((heading, "\n".join(body_lines).strip())) heading = line[3:].strip() body_lines = [] elif heading is not None: body_lines.append(line) if heading is not None: sections.append((heading, "\n".join(body_lines).strip())) return sections def _sections_to_text(sections: list[tuple[str, str]]) -> str: return "".join(f"\n## {h}\n\n{b}\n" for h, b in sections) # --------------------------------------------------------------------------- # Sync implementations # --------------------------------------------------------------------------- def _reminders_list() -> str: p = _reminders_path() if not p.exists() or not p.read_text().strip(): return "No pending reminders." sections = _parse_sections(p.read_text()) if not sections: return "No pending reminders." lines = [] for i, (heading, body) in enumerate(sections, 1): lines.append(f"{i}. {heading}") if body: # Indent body so it reads as belonging to the numbered item for bline in body.splitlines()[:4]: # cap at 4 lines for brevity lines.append(f" {bline}") lines.append("") return "\n".join(lines).rstrip() def _reminders_add(text: str, label: str | None = None) -> str: p = _reminders_path() existing = p.read_text() if p.exists() else "" heading = label or _now_label() section = f"\n## {heading}\n\n{text.strip()}\n" p.write_text(existing.rstrip() + "\n" + section) return f"Reminder added: {heading}" def _reminders_remove(index: int) -> str: p = _reminders_path() if not p.exists() or not p.read_text().strip(): return "No reminders to remove." sections = _parse_sections(p.read_text()) if not sections: return "No reminders to remove." if index < 1 or index > len(sections): return ( f"Index {index} is out of range. " f"There {'is' if len(sections) == 1 else 'are'} {len(sections)} " f"reminder{'s' if len(sections) != 1 else ''} (1–{len(sections)}). " "Call reminders_list to see them." ) removed_heading = sections[index - 1][0] sections.pop(index - 1) p.write_text(_sections_to_text(sections)) return f"Removed reminder {index}: {removed_heading}" def _reminders_clear() -> str: p = _reminders_path() p.write_text("") return "All reminders cleared." # --------------------------------------------------------------------------- # Async wrappers # --------------------------------------------------------------------------- async def reminders_list() -> str: return await asyncio.to_thread(_reminders_list) async def reminders_add(text: str, label: str | None = None) -> str: return await asyncio.to_thread(_reminders_add, text, label) async def reminders_remove(index: int) -> str: return await asyncio.to_thread(_reminders_remove, index) async def reminders_clear() -> str: return await asyncio.to_thread(_reminders_clear)