""" Reminders tools. Reminders are stored in persona/REMINDERS.md and automatically surfaced in the system prompt at Tier 2+. Each reminder can have an optional due date — only due or undated reminders surface in context; future-dated ones are stored but invisible until their date arrives. Operations: reminders_add — append a new reminder, optional due date (YYYY-MM-DD) reminders_list — return all reminders with due status (including future) reminders_remove — remove a single reminder by number reminders_clear — erase all reminders """ import asyncio import re from datetime import datetime, timezone, date as _date 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) def _parse_due(body: str) -> _date | None: """Extract due date from a 'due: YYYY-MM-DD' line in the body, if present.""" m = re.search(r'^due:\s*(\d{4}-\d{2}-\d{2})', body, re.MULTILINE | re.IGNORECASE) if not m: return None try: return _date.fromisoformat(m.group(1)) except ValueError: return None def _today() -> _date: return datetime.now().astimezone().date() def _is_due_or_undated(body: str) -> bool: """Return True if this reminder has no due date or its due date is today or past.""" due = _parse_due(body) return due is None or due <= _today() def _due_label(body: str) -> str: """Return a human-readable due status string for reminders_list output.""" due = _parse_due(body) if due is None: return "" today = _today() if due < today: days = (today - due).days return f" [OVERDUE by {days} day{'s' if days != 1 else ''} — due {due}]" if due == today: return " [due TODAY]" return f" [due: {due}]" def _body_without_due(body: str) -> str: """Strip the due: line from body for display (due status shown in heading line).""" return re.sub(r'^due:\s*\S+\s*\n?', '', body, count=1, flags=re.MULTILINE | re.IGNORECASE).strip() # --------------------------------------------------------------------------- # 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): status = _due_label(body) lines.append(f"{i}. {heading}{status}") display_body = _body_without_due(body) if display_body: for bline in display_body.splitlines()[:4]: lines.append(f" {bline}") lines.append("") return "\n".join(lines).rstrip() def _reminders_add(text: str, label: str | None = None, due: str | None = None) -> str: p = _reminders_path() existing = p.read_text() if p.exists() else "" heading = label or _now_label() body = text.strip() if due: body = f"due: {due}\n{body}" section = f"\n## {heading}\n\n{body}\n" p.write_text(existing.rstrip() + "\n" + section) msg = f"Reminder added: {heading}" if due: msg += f" (due: {due})" return msg 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." # --------------------------------------------------------------------------- # Public helper for context_loader # --------------------------------------------------------------------------- def load_due_reminders() -> str: """Return REMINDERS.md content filtered to only due and undated sections. Called by context_loader at Tier 2+. Future-dated reminders are excluded from the system prompt until their due date arrives. """ p = _reminders_path() if not p.exists(): return "" text = p.read_text() if not text.strip(): return "" sections = _parse_sections(text) due_sections = [(h, b) for h, b in sections if _is_due_or_undated(b)] if not due_sections: return "" # Strip the raw due: line from body — the date is already part of the heading context cleaned = [(h, _body_without_due(b)) for h, b in due_sections] return _sections_to_text(cleaned).strip() # --------------------------------------------------------------------------- # Async wrappers # --------------------------------------------------------------------------- async def reminders_list() -> str: return await asyncio.to_thread(_reminders_list) async def reminders_add(text: str, label: str | None = None, due: str | None = None) -> str: return await asyncio.to_thread(_reminders_add, text, label, due) 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 context at the start of each session (Tier 2+). " "Use this when the user asks you to remember something or follow up on something. " "Set a due date to suppress the reminder until that date — useful for future tasks " "that would be noise today." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "text": types.Schema(type=types.Type.STRING, description="The reminder text"), "label": types.Schema(type=types.Type.STRING, description="Optional heading (e.g. 'Follow up on NC Talk'). Defaults to current timestamp."), "due": types.Schema(type=types.Type.STRING, description="Optional due date in YYYY-MM-DD format. Reminder is hidden from context until this date arrives. Omit for an always-visible reminder."), }, required=["text"], ), ), types.FunctionDeclaration( name="reminders_list", description=( "Read all pending reminders, including future-dated ones not yet in context. " "Shows due status for each (due today, overdue, or future date). " "Use this before adding to avoid duplicates, or to show the user what's queued." ), 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 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 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={}), ), ]