""" 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 google.genai import types 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) DECLARATIONS = [ types.FunctionDeclaration( name="reminders_add", description=( "Add a new reminder to REMINDERS.md. Reminders are automatically surfaced " "in your context at the start of each session (Tier 2+). " "Use this when the user asks you to remember something, follow up on something, " "or surface a note at the next session." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "text": types.Schema(type=types.Type.STRING, description="The reminder text to add"), "label": types.Schema(type=types.Type.STRING, description="Optional heading for this reminder (e.g. 'Follow up on NC Talk'). Defaults to current timestamp."), }, required=["text"], ), ), types.FunctionDeclaration( name="reminders_list", description=( "Read all current pending reminders from REMINDERS.md. " "Use this to check what reminders are queued before adding duplicates, " "or to show the user what's pending." ), parameters=types.Schema(type=types.Type.OBJECT, properties={}), ), types.FunctionDeclaration( name="reminders_remove", description=( "Remove a single reminder by its number. " "Call reminders_list first to get the numbered list, then pass the number of the reminder to remove." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "index": types.Schema(type=types.Type.INTEGER, description="The number of the reminder to remove (1 = first item in reminders_list output)."), }, required=["index"], ), ), types.FunctionDeclaration( name="reminders_clear", description=( "Erase all pending reminders from REMINDERS.md. " "Use this after you have acknowledged and acted on the reminders shown in your context." ), parameters=types.Schema(type=types.Type.OBJECT, properties={}), ), ]