"""Built-in tool: schedule a delayed recall for the current session.""" from __future__ import annotations from datetime import datetime, timezone from ._internal.base import Tool, ToolResult, current_session_id, current_user_role from ._internal.time_parser import parse_when class ScheduleRecallTool(Tool): name = "schedule_recall" description = ( "Schedule a headless callback for the current session. " "At the chosen time Navi wakes up, reads your self-instruction, and continues working using other tools. " "Only one pending recall per session is allowed — cancel the old one first if you need a new timer.\n\n" "CORE PRINCIPLE: this is a TOOL for CONTINUING WORK, not a chat reminder.\n" " BAD message: 'Tell the user that 2 hours have passed.'\n" " GOOD message: 'Read /tmp/build.log with filesystem. If errors, read last 50 lines and report. Otherwise confirm success.'\n\n" "Call types:\n" " once — single delayed action (check logs in 30m, continue after reboot).\n" " recurring — periodic action with interval_seconds (poll API every 5 min, check inbox every 15 min).\n" " immediate — fire ASAP; use to offload heavy multi-tool work without blocking the chat.\n\n" "Popular scenarios:\n" " 1. Hit iteration limit? Schedule immediate with context 'Continue Nginx config from step 3...'\n" " 2. Waiting for a build/export? Schedule once for estimated finish time and tell yourself which files to check.\n" " 3. Periodic monitoring? Schedule recurring 900s (15 min) with context 'Read /var/log/app/errors.log...'\n" " 4. Heavy task with 20+ tool calls? Use immediate so user can keep chatting while you work headlessly.\n\n" "Rules:\n" " • additional_context_message must read like a todo item for yourself.\n" " • Mention specific tools, files, or URLs — future-you has the same tools but not your short-term memory.\n" " • Multi-phase tasks: chain recalls — phase 1 runs, then schedules recall for phase 2.\n" " • Only one pending recall per session. Use manage_recall cancel before scheduling a new one." ) parameters = { "type": "object", "properties": { "call_type": { "type": "string", "enum": ["once", "recurring", "immediate"], "description": "Type of recall: once (single), recurring (repeats), immediate (ASAP).", }, "when": { "type": "string", "description": ( "When to trigger. Absolute ISO datetime or relative phrase " "like '30m', '2h 15m', '1d 6h', 'in 3 hours', 'tomorrow at 09:00'. " "Ignored for immediate calls." ), }, "interval_seconds": { "type": "integer", "description": "Repeat interval in seconds. Required only for recurring.", }, "internal_comment": { "type": "string", "description": "Optional human-readable note why this recall was set.", }, "additional_context_message": { "type": "string", "description": ( "Self-instruction describing exactly what to do when the recall fires. " "Write it as a todo for yourself: mention specific tools, files, or steps. " "This is injected as a system message and Navi acts on it with other tools." ), }, }, "required": ["call_type", "additional_context_message"], } def __init__(self, scheduler: "RecallScheduler" | None = None) -> None: self._scheduler = scheduler async def execute(self, params: dict) -> ToolResult: from navi.core.scheduler import RecallExistsError, RecallScheduler scheduler = self._scheduler if scheduler is None: return ToolResult( success=False, output="Scheduler not available.", error="no scheduler", ) session_id = current_session_id.get(None) if not session_id: return ToolResult( success=False, output="No current session.", error="missing session_id", ) call_type = (params.get("call_type") or "").strip().lower() when = (params.get("when") or "").strip() interval_seconds: int | None = params.get("interval_seconds") internal_comment: str | None = params.get("internal_comment") additional_context_message = (params.get("additional_context_message") or "").strip() if call_type not in ("once", "recurring", "immediate"): return ToolResult( success=False, output=f"Invalid call_type: {call_type}. Use once, recurring, or immediate.", error="bad call_type", ) if call_type == "recurring": if not interval_seconds or interval_seconds <= 0: return ToolResult( success=False, output="interval_seconds is required and must be > 0 for recurring recalls.", error="missing interval", ) else: interval_seconds = None # Resolve trigger time if call_type == "immediate": trigger_at = datetime.now(timezone.utc) elif when: trigger_at, error = parse_when(when) if error: return ToolResult(success=False, output=error, error="parse_failed") else: return ToolResult( success=False, output="'when' is required for once/recurring recalls.", error="missing when", ) if not additional_context_message: return ToolResult( success=False, output="additional_context_message is required.", error="missing context", ) try: recall = await scheduler.schedule_recall( session_id=session_id, call_type=call_type, trigger_at=trigger_at, interval_seconds=interval_seconds, internal_comment=internal_comment, additional_context_message=additional_context_message, ) except RecallExistsError as exc: return ToolResult( success=False, output=f"This session already has a pending recall. Cancel it first: {exc}", error="recall_exists", ) except Exception as exc: return ToolResult( success=False, output=f"Failed to schedule recall: {exc}", error="schedule_failed", ) return ToolResult( success=True, output=( f"Recall scheduled.\n" f" ID: {recall.id}\n" f" Type: {recall.call_type}\n" f" Trigger: {recall.trigger_at.isoformat()}\n" + ( f" Interval: {recall.interval_seconds}s\n" if recall.interval_seconds else "" ) + (f" Comment: {recall.internal_comment}\n" if recall.internal_comment else "") ), )