brain/tools/server.py

"""Tools that run in the brain: search, weather, reminders, memory. Blueprint 8.1 and 8.2."""

from datetime import datetime, timedelta, tzinfo
from typing import Any

import httpx

from ..memory import service as memory
from ..memory.scope import forced_scope
from ..persona.types import Mode
from . import weather
from .registry import Tool, ToolContext, ToolOutcome, failed, ok, schema, string

DEFAULT_CITY = "Kolkata"

#: A reminder "in the past" by less than this is treated as "now" — the model computed the time
#: a few seconds before the call landed.
PAST_TOLERANCE = timedelta(minutes=2)


async def _web_search(args: dict[str, Any], ctx: ToolContext) -> ToolOutcome:
    """Blueprint 8.2: a separate grounded call; the sources become chips under the reply."""
    answer, sources = await ctx.llm.grounded_search(str(args["query"]))
    if not answer:
        return ToolOutcome(failed("the search found nothing useful"))
    return ToolOutcome(ok("searched", answer=answer), sources=sources)


async def _get_weather(args: dict[str, Any], ctx: ToolContext) -> ToolOutcome:
    place = (
        str(args.get("location") or "").strip()
        or str(ctx.hud.get("city") or "").strip()
        or str((ctx.settings.get("hud") or {}).get("city") or "").strip()
        or DEFAULT_CITY
    )
    client = ctx.http or httpx.AsyncClient(timeout=10.0)
    try:
        report = await weather.current(client, place)
    except weather.WeatherError as exc:
        return ToolOutcome(failed(str(exc)))
    finally:
        if ctx.http is None:
            await client.aclose()
    return ToolOutcome(ok("weather", **report))


def parse_when(raw: str, ctx: ToolContext) -> datetime | None:
    """An ISO time from the model. A naive one is read in HIS timezone, never UTC (CLAUDE.md)."""
    try:
        when = datetime.fromisoformat(raw.strip().replace("Z", "+00:00"))
    except ValueError:
        return None
    if when.tzinfo is None:
        when = when.replace(tzinfo=ctx.tz)
    return when


def local_label(when: datetime, tz: tzinfo) -> str:
    local = when.astimezone(tz)
    hour = local.hour % 12 or 12
    suffix = "AM" if local.hour < 12 else "PM"
    return f"{local.strftime('%a %d %b')}, {hour}:{local.minute:02d} {suffix}"


async def _set_reminder(args: dict[str, Any], ctx: ToolContext) -> ToolOutcome:
    when = parse_when(str(args["when_iso"]), ctx)
    if when is None:
        return ToolOutcome(failed("when_iso must be an ISO 8601 date and time"))
    if when < ctx.now - PAST_TOLERANCE:
        return ToolOutcome(failed("that time has already passed"))
    when = max(when, ctx.now)

    text = str(args["text"]).strip()
    row = await ctx.store.add_reminder(ctx.user_id, ctx.mode, text, when)
    return ToolOutcome(ok("reminder set", id=str(row["id"]), due=local_label(when, ctx.tz)))


async def _list_reminders(args: dict[str, Any], ctx: ToolContext) -> ToolOutcome:
    rows = await ctx.store.list_reminders(ctx.user_id)
    # Normal mode never lists a reminder that was created in GF mode — its text may be private.
    visible = [row for row in rows if ctx.mode is Mode.GF or row["mode"] == Mode.NORMAL.value]
    return ToolOutcome(
        ok(
            f"{len(visible)} upcoming",
            reminders=[
                {
                    "id": str(row["id"]),
                    "text": row["text"],
                    "due": local_label(row["due_at"], ctx.tz),
                }
                for row in visible
            ],
        )
    )


async def _cancel_reminder(args: dict[str, Any], ctx: ToolContext) -> ToolOutcome:
    cancelled = await ctx.store.cancel_reminder(ctx.user_id, str(args["id"]))
    return ToolOutcome(ok("cancelled") if cancelled else failed("no reminder with that id"))


async def _remember(args: dict[str, Any], ctx: ToolContext) -> ToolOutcome:
    scope = forced_scope(args.get("scope"), ctx.mode)
    row, duplicate = await memory.remember(
        ctx.store, ctx.llm, ctx.user_id, str(args["fact"]), scope, is_manual=True
    )
    if duplicate is not None:
        return ToolOutcome(ok("already remembered", existing=duplicate.content))
    return ToolOutcome(ok("remembered", scope=scope.value))


async def _forget(args: dict[str, Any], ctx: ToolContext) -> ToolOutcome:
    """Finds the memory but never deletes it: he confirms on the device first (blueprint 8.1)."""
    hit = await memory.find_to_forget(ctx.store, ctx.llm, ctx.user_id, str(args["query"]), ctx.mode)
    if hit is None:
        return ToolOutcome(failed("nothing like that is in memory"))
    action = {"type": "confirm_forget", "memoryId": hit.id, "content": hit.content}
    return ToolOutcome(ok("asked him to confirm", found=hit.content), client_actions=[action])


SERVER_TOOLS: list[Tool] = [
    Tool(
        "web_search",
        "Search the web and get a current answer with sources — news, scores, prices, anything "
        "that changes. Use it instead of guessing about recent events.",
        schema({"query": string("What to find out")}, ["query"]),
        _web_search,
    ),
    Tool(
        "get_weather",
        "Current weather and today's range. Omit location to use where he is.",
        schema({"location": string("A city or place name")}),
        _get_weather,
    ),
    Tool(
        "set_reminder",
        "Set a reminder that arrives as a notification at a specific time. Work out the exact "
        "time from CURRENT TIME in your instructions and his timezone.",
        schema(
            {
                "text": string("What to remind him about, as he should read it"),
                "when_iso": string("ISO 8601 local date and time, e.g. 2026-09-19T14:45:00"),
            },
            ["text", "when_iso"],
        ),
        _set_reminder,
    ),
    Tool(
        "list_reminders",
        "List his upcoming reminders.",
        schema({}),
        _list_reminders,
    ),
    Tool(
        "cancel_reminder",
        "Cancel a reminder by its id (get ids from list_reminders).",
        schema({"id": string("The reminder id")}, ["id"]),
        _cancel_reminder,
    ),
    Tool(
        "remember",
        "Save a fact about him for the long term, when he explicitly asks you to remember "
        "something. One short third-person sentence in English.",
        schema(
            {
                "fact": string("The fact, e.g. 'His dentist appointment is on Friday.'"),
                "scope": string(
                    "private for anything romantic or intimate; shared for everything else",
                    ["shared", "private"],
                ),
            },
            ["fact"],
        ),
        _remember,
    ),
    Tool(
        "forget",
        "Find a memory he wants deleted. He confirms on his device before anything is deleted.",
        schema({"query": string("What he wants forgotten")}, ["query"]),
        _forget,
    ),
]