Add admin-only Homebox MCP connector

Load an allowlisted subset of homebox-mcp tools (read + safe writes, no
deletes) over stdio and compile them into a separate admin agent, so
non-admins never see them. Each tool also re-checks is_admin per call.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

.env.example

@@ -10,6 +10,12 @@ POSTGRES_DB=raggr
 # JWT Configuration
 JWT_SECRET_KEY=your-secret-key-here
 
+# Metrics Configuration
+# Bearer token for the /metrics endpoint (Prometheus scrape + app token gate).
+# Generate with: openssl rand -hex 32
+# Leave empty to keep /metrics returning 403 (fail-closed).
+METRICS_TOKEN=
+
 # Paperless Configuration
 PAPERLESS_TOKEN=your-paperless-token
 BASE_URL=192.168.1.5:8000
@@ -18,6 +24,7 @@ BASE_URL=192.168.1.5:8000
 # If set, uses llama-server as the primary LLM backend with OpenAI as fallback
 LLAMA_SERVER_URL=http://192.168.1.213:8080/v1
 LLAMA_MODEL_NAME=llama-3.1-8b-instruct
+TTFT_TIMEOUT_SECONDS=30
 
 # Embedding Server Configuration
 # If set, uses a custom OpenAI-compatible embedding server (e.g. llama-server)
@@ -107,11 +114,21 @@ ALLOWED_IMESSAGE_NUMBERS=
 # IMESSAGE_RATE_LIMIT_MAX=10
 # IMESSAGE_RATE_LIMIT_WINDOW=60
 
-# Google Calendar Configuration (via gws CLI)
-GOOGLE_CALENDAR_ENABLED=true
-# Export credentials: gws auth export --unmasked > credentials.json
-# The file is mounted into the container at /app/config/gws-credentials.json
-GOOGLE_WORKSPACE_CLI_CREDENTIALS_FILE=/app/config/gws-credentials.json
+# Google Calendar + Gmail Configuration (Python SDK)
+# Run scripts/google_auth.py to generate token.json from your OAuth2 credentials:
+#   python scripts/google_auth.py --credentials credentials.json --token token.json
+# Then mount token.json into the container (see docker-compose.yml).
+# If GOOGLE_TOKEN_FILE points to a valid token, both Calendar and Gmail are enabled.
+GOOGLE_TOKEN_FILE=/app/config/google-token.json
+
+# Cronicle Integration
+# Shared secret for Cronicle to authenticate cron job trigger requests.
+# Generate with: openssl rand -hex 32
+CRON_SECRET=
+# Cronicle server URL and API key (from Cronicle Admin → API Keys)
+# Used to auto-register scheduled jobs on startup.
+CRONICLE_URL=http://cronicle:3012
+CRONICLE_API_KEY=
 
 # SimbaKit / PetKit Configuration
 # PetKit account credentials (used by SimbaKit to poll device data)
@@ -122,6 +139,11 @@ PETKIT_TIMEZONE=America/New_York
 # Internal URL for SimbaKit service (Docker network)
 SIMBAKIT_URL=http://simbakit:8585
 
+# Homebox inventory (admin-only agent tools via homebox-mcp, needs Homebox >= 0.26)
+# API key from Homebox Profile -> API Keys (starts with hb_). Leave empty to disable.
+HOMEBOX_URL=https://homebox.example.com
+HOMEBOX_TOKEN=
+
 # Obsidian Configuration (headless sync)
 # Obsidian account credentials (used for `ob login` on container startup)
 OBSIDIAN_EMAIL=your-obsidian-email
@@ -136,3 +158,11 @@ OBSIDIAN_DEVICE_NAME=simbarag
 OBSIDIAN_CONTINUOUS_SYNC=false
 # Local path to Obsidian vault (where files are synced)
 OBSIDIAN_VAULT_PATH=/app/data/obsidian
+
+# LangGraph checkpoint store (SQLite). Must be on a persistent volume.
+# Both blue and green containers share this volume via docker-compose.
+CHECKPOINT_DB_PATH=/app/checkpoints/checkpoints.db
+
+# Beli restaurant recommendations
+# Beli account credentials for each profile are stored encrypted in the database.
+# No additional env vars needed — auth is done via the Beli API at login time.

.gitignore

@@ -12,6 +12,7 @@ wheels/
 # Environment files
 .env
 credentials.json
+token.json
 
 # Database files
 database/

CLAUDE.md

@@ -89,6 +89,7 @@ make redeploy    # git pull + scripts/deploy.sh (build new color, wait healthy,
   - `image_process.py` - Image description with LLM
   - `request.py` - Paperless-NGX API client
   - `obsidian_service.py`, `mealie_service.py`, `ynab_service.py`, `s3_client.py`, `simbakit_service.py` - integrations with Obsidian notes, Mealie recipes, YNAB budgets, S3, and SimbaKit
+  - `homebox_mcp.py` - Homebox inventory tools via the `homebox-mcp` stdio MCP server; admin-only (compiled into `app.admin_agent`, picked per user by `get_agent_for`)
 - `scripts/` - Administrative and utility scripts
   - `add_user.py` - Create users manually
   - `user_message_stats.py` - User message statistics
@@ -155,7 +156,7 @@ See `.env.example`. Key ones:
 - `TWILIO_*`, `ALLOWED_IMESSAGE_NUMBERS`, `ALLOWED_WHATSAPP_NUMBERS` - iMessage/WhatsApp webhooks
 - `MAILGUN_*`, `EMAIL_HMAC_SECRET`, `FERNET_KEY` - email integration
 - `SENDBLUE_*` - SMS sending for scheduled messages
-- `YNAB_*`, `MEALIE_*`, `OBSIDIAN_*`, `IMMICH_*`, `SIMBAKIT_URL` - household integrations
+- `YNAB_*`, `MEALIE_*`, `OBSIDIAN_*`, `IMMICH_*`, `SIMBAKIT_URL`, `HOMEBOX_URL`/`HOMEBOX_TOKEN` - household integrations
 
 ## Documentation
 

Dockerfile

@@ -15,6 +15,9 @@ RUN apt-get update && apt-get install -y \
 # Add uv to PATH
 ENV PATH="/root/.local/bin:$PATH"
 
+# Homebox MCP server (stdio), pinned so the tool surface can't drift
+RUN uv tool install homebox-mcp==1.1.0
+
 # Copy dependency files
 COPY pyproject.toml ./
 

Makefile

@@ -55,8 +55,7 @@ migrate:
 	$(COMPOSE) exec $(RAGGR) aerich upgrade
 
 migrate-new:
-	@read -p "Migration name: " name; \
-	$(COMPOSE) exec $(RAGGR) aerich migrate --name $$name
+	@./scripts/migrate-new.sh
 
 migrate-history:
 	$(COMPOSE) exec $(RAGGR) aerich history

app.py

@@ -1,10 +1,18 @@
 import asyncio
+import hmac
 import logging
 import os
 from datetime import timedelta
 
 from dotenv import load_dotenv
-from quart import Quart, jsonify, render_template, send_from_directory
+from quart import (
+    Quart,
+    Response,
+    jsonify,
+    render_template,
+    request,
+    send_from_directory,
+)
 from quart_jwt_extended import JWTManager, get_jwt_identity, jwt_refresh_token_required
 from tortoise import Tortoise
 
@@ -16,8 +24,13 @@ import blueprints.users
 import blueprints.whatsapp
 import blueprints.imessage
 import blueprints.scheduled_messages
+import blueprints.cron
+import blueprints.beli
 import blueprints.users.models
 from config.db import TORTOISE_CONFIG
+from metrics import render_metrics
+from blueprints.conversation.agents import build_agent
+from utils.homebox_mcp import load_homebox_tools
 
 # Load environment variables
 load_dotenv()
@@ -54,6 +67,8 @@ app.register_blueprint(blueprints.rag.rag_blueprint)
 app.register_blueprint(blueprints.whatsapp.whatsapp_blueprint)
 app.register_blueprint(blueprints.imessage.imessage_blueprint)
 app.register_blueprint(blueprints.scheduled_messages.scheduled_messages_blueprint)
+app.register_blueprint(blueprints.cron.cron_blueprint)
+app.register_blueprint(blueprints.beli.beli_blueprint)
 
 
 async def _obsidian_sync_loop():
@@ -84,19 +99,34 @@ async def lifespan():
     await Tortoise.init(config=TORTOISE_CONFIG)
     logging.info("Tortoise ORM initialized successfully")
 
-    watcher_task = None
-    if os.getenv("OBSIDIAN_CONTINUOUS_SYNC") == "true":
-        watcher_task = asyncio.create_task(_obsidian_sync_loop())
+    import aiosqlite
+    from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
 
-    from blueprints.scheduled_messages.scheduler import scheduled_messages_loop
+    checkpoint_path = os.getenv("CHECKPOINT_DB_PATH", "/app/checkpoints/checkpoints.db")
+    async with aiosqlite.connect(checkpoint_path) as conn:
+        checkpointer = AsyncSqliteSaver(conn)
+        await checkpointer.setup()
+        app.main_agent = build_agent(checkpointer)
+        homebox_tools = await load_homebox_tools()
+        app.admin_agent = (
+            build_agent(checkpointer, homebox_tools)
+            if homebox_tools
+            else app.main_agent
+        )
+        logging.info(f"Agent checkpointer ready at {checkpoint_path}")
 
-    scheduler_task = asyncio.create_task(scheduled_messages_loop())
+        from utils.cronicle import ensure_beli_nightly_event
 
-    yield
+        await ensure_beli_nightly_event()
 
-    scheduler_task.cancel()
-    if watcher_task is not None:
-        watcher_task.cancel()
+        watcher_task = None
+        if os.getenv("OBSIDIAN_CONTINUOUS_SYNC") == "true":
+            watcher_task = asyncio.create_task(_obsidian_sync_loop())
+
+        yield
+
+        if watcher_task is not None:
+            watcher_task.cancel()
 
     logging.info("Closing Tortoise ORM connections...")
     await Tortoise.close_connections()
@@ -129,6 +159,27 @@ ALLOWED_STATIC_EXTENSIONS = {
 ALLOWED_JSON_FILES = {"manifest.json"}
 
 
+# Prometheus metrics endpoint. Fail-closed: 403 unless a valid bearer token
+# matching METRICS_TOKEN is presented. Registered before the catch-all SPA
+# route so it is never shadowed.
+@app.get("/metrics")
+async def metrics_endpoint():
+    expected = os.environ.get("METRICS_TOKEN", "")
+    if not expected:
+        return "unauthorized", 403
+    auth = request.headers.get("Authorization", "")
+    if not auth.startswith("Bearer "):
+        return "unauthorized", 403
+    provided = auth[len("Bearer ") :].strip()
+    if not hmac.compare_digest(provided.encode(), expected.encode()):
+        return "unauthorized", 403
+    return Response(
+        render_metrics(),
+        status=200,
+        content_type="text/plain; version=0.0.4",
+    )
+
+
 # Serve the React app for all routes (catch-all)
 @app.route("/", defaults={"path": ""})
 @app.route("/<path:path>")

blueprints/beli/__init__.py

@@ -0,0 +1,185 @@
+import hmac
+import logging
+import os
+
+from quart import Blueprint, jsonify, request
+from quart_jwt_extended import get_jwt_identity, jwt_refresh_token_required
+
+from .logic import (
+    MAX_PROFILES_PER_USER,
+    _get_access_token,
+    _refresh_friend_sample,
+    refresh_profile,
+    schedule_nightly_refresh,
+)
+from .models import BeliProfile, BeliRestaurant, BeliStatus
+
+beli_blueprint = Blueprint("beli", __name__, url_prefix="/api/beli")
+logger = logging.getLogger(__name__)
+
+
+def _check_cron_secret() -> bool:
+    expected = os.environ.get("CRON_SECRET", "")
+    if not expected:
+        return False
+    provided = request.headers.get("X-Cron-Secret", "")
+    return hmac.compare_digest(provided.encode(), expected.encode())
+
+
+def _serialize_profile(profile: BeliProfile) -> dict:
+    return {
+        "id": str(profile.id),
+        "beli_email": profile.beli_email,
+        "beli_user_id": profile.beli_user_id,
+        "enabled": profile.enabled,
+        "last_refreshed_at": (
+            profile.last_refreshed_at.isoformat() if profile.last_refreshed_at else None
+        ),
+        "last_error": profile.last_error,
+        "created_at": profile.created_at.isoformat(),
+    }
+
+
+def _serialize_restaurant(r: BeliRestaurant) -> dict:
+    return {
+        "id": str(r.id),
+        "profile_id": str(r.profile_id),
+        "beli_business_id": r.beli_business_id,
+        "name": r.name,
+        "city": r.city,
+        "neighborhood": r.neighborhood,
+        "cuisine": r.cuisine,
+        "price_range": r.price_range,
+        "status": r.status.value,
+        "rank_index": r.rank_index,
+        "score": float(r.score) if r.score is not None else None,
+        "notes": r.notes,
+        "tags": r.tags,
+        "fetched_at": r.fetched_at.isoformat(),
+    }
+
+
+@beli_blueprint.route("/profiles", methods=["GET"])
+@jwt_refresh_token_required
+async def list_profiles():
+    user_id = get_jwt_identity()
+    profiles = await BeliProfile.filter(user_id=user_id).order_by("created_at").all()
+    return jsonify([_serialize_profile(p) for p in profiles])
+
+
+@beli_blueprint.route("/profiles", methods=["POST"])
+@jwt_refresh_token_required
+async def create_profile():
+    user_id = get_jwt_identity()
+
+    count = await BeliProfile.filter(user_id=user_id).count()
+    if count >= MAX_PROFILES_PER_USER:
+        return (
+            jsonify(
+                {"error": f"Maximum {MAX_PROFILES_PER_USER} Beli profiles per user"}
+            ),
+            400,
+        )
+
+    data = await request.get_json()
+    if not data:
+        return jsonify({"error": "Invalid payload"}), 400
+
+    email = (data.get("beli_email") or "").strip()
+    password = (data.get("beli_password") or "").strip()
+    if not email or not password:
+        return jsonify({"error": "beli_email and beli_password are required"}), 400
+
+    if await BeliProfile.exists(user_id=user_id, beli_email=email):
+        return jsonify({"error": "A profile with this email already exists"}), 409
+
+    from utils.beli_service import authenticate
+
+    try:
+        session = await authenticate(email, password)
+    except Exception as e:
+        return jsonify({"error": f"Beli authentication failed: {e}"}), 400
+
+    profile = await BeliProfile.create(
+        user_id=user_id,
+        beli_email=email,
+        beli_password=password,
+        beli_user_id=session["user_id"],
+        beli_refresh_token=session["refresh_token"],
+        enabled=True,
+    )
+    return jsonify(_serialize_profile(profile)), 201
+
+
+@beli_blueprint.route("/profiles/<profile_id>", methods=["DELETE"])
+@jwt_refresh_token_required
+async def delete_profile(profile_id: str):
+    user_id = get_jwt_identity()
+    profile = await BeliProfile.get_or_none(id=profile_id, user_id=user_id)
+    if not profile:
+        return jsonify({"error": "Not found"}), 404
+
+    await BeliRestaurant.filter(profile=profile).delete()
+    await profile.delete()
+    return jsonify({"status": "deleted"})
+
+
+@beli_blueprint.route("/profiles/<profile_id>/refresh", methods=["POST"])
+@jwt_refresh_token_required
+async def refresh_profile_endpoint(profile_id: str):
+    user_id = get_jwt_identity()
+    profile = await BeliProfile.get_or_none(id=profile_id, user_id=user_id)
+    if not profile:
+        return jsonify({"error": "Not found"}), 404
+
+    result = await refresh_profile(str(profile.id))
+    return jsonify(result)
+
+
+@beli_blueprint.route("/restaurants", methods=["GET"])
+@jwt_refresh_token_required
+async def list_restaurants():
+    user_id = get_jwt_identity()
+    profile_ids = [
+        p.id for p in await BeliProfile.filter(user_id=user_id).only("id").all()
+    ]
+
+    qs = BeliRestaurant.filter(profile_id__in=profile_ids)
+
+    status_filter = request.args.get("status")
+    city_filter = request.args.get("city")
+    if status_filter:
+        try:
+            qs = qs.filter(status=BeliStatus(status_filter))
+        except ValueError:
+            return jsonify({"error": f"Invalid status: {status_filter}"}), 400
+    if city_filter:
+        qs = qs.filter(city__icontains=city_filter)
+
+    restaurants = await qs.order_by("rank_index", "name").all()
+    return jsonify([_serialize_restaurant(r) for r in restaurants])
+
+
+@beli_blueprint.route("/profiles/<profile_id>/refresh-friends", methods=["POST"])
+@jwt_refresh_token_required
+async def refresh_friends_endpoint(profile_id: str):
+    user_id = get_jwt_identity()
+    profile = await BeliProfile.get_or_none(id=profile_id, user_id=user_id)
+    if not profile:
+        return jsonify({"error": "Not found"}), 404
+
+    try:
+        access_token = await _get_access_token(profile)
+        upserted = await _refresh_friend_sample(profile, access_token)
+        return jsonify({"ok": True, "upserted": upserted})
+    except Exception as e:
+        return jsonify({"error": str(e)}), 500
+
+
+@beli_blueprint.route("/cron/refresh", methods=["POST"])
+async def cron_nightly_refresh():
+    if not _check_cron_secret():
+        return jsonify({"error": "Unauthorized"}), 401
+
+    result = await schedule_nightly_refresh()
+    return jsonify(result)

blueprints/beli/logic.py

@@ -0,0 +1,252 @@
+import asyncio
+import logging
+import random
+from datetime import datetime, timezone
+
+from utils.beli_service import (
+    RANKING_CATEGORIES,
+    REQUEST_SPACING,
+    authenticate,
+    extract_business,
+    fetch_bookmarks,
+    fetch_following,
+    fetch_user_rankings,
+    is_token_expired,
+    refresh_access_token,
+)
+
+from tortoise.functions import Max
+
+from .models import BeliProfile, BeliRestaurant, BeliStatus
+
+logger = logging.getLogger(__name__)
+
+MAX_PROFILES_PER_USER = 3
+NIGHTLY_BATCH_SIZE = 3
+NIGHTLY_WINDOW_SECONDS = 4 * 3600  # spread refreshes across a 4-hour window
+FRIEND_SAMPLE_SIZE = 5  # friends to randomly refresh per profile per night
+
+
+async def _get_access_token(profile: BeliProfile) -> str:
+    """Return a live access token, refreshing or re-authenticating as needed."""
+    if profile.beli_refresh_token and not is_token_expired(profile.beli_refresh_token):
+        try:
+            token = await refresh_access_token(profile.beli_refresh_token)
+            return token
+        except Exception:
+            logger.warning(
+                "Refresh token failed for profile %s, falling back to password",
+                profile.id,
+            )
+
+    session = await authenticate(profile.beli_email, profile.beli_password)
+    profile.beli_user_id = session["user_id"]
+    profile.beli_refresh_token = session["refresh_token"]
+    await profile.save()
+    return session["access_token"]
+
+
+async def _refresh_friend_sample(profile: BeliProfile, access_token: str) -> int:
+    """Fetch rankings for a random sample of followed users and upsert them."""
+    try:
+        following = await fetch_following(access_token, profile.beli_user_id)
+    except Exception as e:
+        logger.debug("Beli following fetch skipped for profile %s: %s", profile.id, e)
+        return 0
+
+    if not following:
+        return 0
+
+    # Pick the FRIEND_SAMPLE_SIZE stalest friends — never-fetched first, then oldest
+    fetched_rows = (
+        await BeliRestaurant.filter(profile=profile)
+        .exclude(friend_beli_user_id="")
+        .group_by("friend_beli_user_id")
+        .annotate(last_fetched=Max("fetched_at"))
+        .values("friend_beli_user_id", "last_fetched")
+    )
+    last_fetched: dict[str, datetime] = {
+        r["friend_beli_user_id"]: r["last_fetched"] for r in fetched_rows
+    }
+    _epoch = datetime.fromtimestamp(0, tz=timezone.utc)
+    sample = sorted(
+        following,
+        key=lambda f: last_fetched.get(str(f.get("id") or ""), _epoch),
+    )[:FRIEND_SAMPLE_SIZE]
+    upserted = 0
+
+    for friend in sample:
+        friend_id = str(friend.get("id") or "")
+        friend_name = friend.get("full_name") or friend.get("username") or friend_id
+        if not friend_id:
+            continue
+
+        for category in RANKING_CATEGORIES:
+            await asyncio.sleep(REQUEST_SPACING)
+            try:
+                results = await fetch_user_rankings(access_token, friend_id, category)
+            except Exception as e:
+                # 405 = private profile; log quietly and skip
+                logger.debug(
+                    "Beli friend ranking skipped (%s/%s): %s", friend_id, category, e
+                )
+                continue
+
+            for idx, item in enumerate(results):
+                biz = extract_business(item, rank_index=idx + 1)
+                if not biz:
+                    continue
+                await BeliRestaurant.update_or_create(
+                    profile=profile,
+                    beli_business_id=biz["beli_business_id"],
+                    status=BeliStatus.BEEN,
+                    friend_beli_user_id=friend_id,
+                    beli_category=category,
+                    defaults={
+                        "friend_name": friend_name,
+                        "name": biz["name"],
+                        "city": biz["city"],
+                        "neighborhood": biz["neighborhood"],
+                        "cuisine": biz["cuisine"],
+                        "price_range": biz["price_range"],
+                        "rank_index": biz["rank_index"],
+                        "score": biz["score"],
+                        "notes": None,
+                        "tags": biz["tags"],
+                    },
+                )
+                upserted += 1
+
+    if upserted:
+        logger.info(
+            "Beli profile %s: %d friend restaurants upserted from %d friends",
+            profile.id,
+            upserted,
+            len(sample),
+        )
+    return upserted
+
+
+async def refresh_profile(profile_id: str) -> dict:
+    """Fetch the latest rankings and bookmarks for a profile and upsert them."""
+    profile = await BeliProfile.get_or_none(id=profile_id)
+    if not profile or not profile.enabled:
+        return {"skipped": True}
+
+    try:
+        access_token = await _get_access_token(profile)
+
+        upserted = 0
+
+        # Fetch all ranking categories (restaurants the user has been to and ranked)
+        for category in RANKING_CATEGORIES:
+            await asyncio.sleep(REQUEST_SPACING)
+            results = await fetch_user_rankings(
+                access_token, profile.beli_user_id, category
+            )
+            for idx, item in enumerate(results):
+                biz = extract_business(item, rank_index=idx + 1)
+                if not biz:
+                    continue
+                await BeliRestaurant.update_or_create(
+                    profile=profile,
+                    beli_business_id=biz["beli_business_id"],
+                    status=BeliStatus.BEEN,
+                    friend_beli_user_id="",
+                    beli_category=category,
+                    defaults={
+                        "name": biz["name"],
+                        "city": biz["city"],
+                        "neighborhood": biz["neighborhood"],
+                        "cuisine": biz["cuisine"],
+                        "price_range": biz["price_range"],
+                        "rank_index": biz["rank_index"],
+                        "score": biz["score"],
+                        "notes": biz["notes"],
+                        "tags": biz["tags"],
+                    },
+                )
+                upserted += 1
+
+        # Fetch bookmarks (want-to-try list) — non-fatal if unavailable
+        await asyncio.sleep(REQUEST_SPACING)
+        try:
+            bookmarks = await fetch_bookmarks(access_token)
+        except Exception as e:
+            logger.warning(
+                "Beli bookmarks fetch failed for profile %s: %s", profile_id, e
+            )
+            bookmarks = []
+        for item in bookmarks:
+            biz = extract_business(item)
+            if not biz:
+                continue
+            await BeliRestaurant.update_or_create(
+                profile=profile,
+                beli_business_id=biz["beli_business_id"],
+                status=BeliStatus.WANT_TO_TRY,
+                friend_beli_user_id="",
+                beli_category="",
+                defaults={
+                    "name": biz["name"],
+                    "city": biz["city"],
+                    "neighborhood": biz["neighborhood"],
+                    "cuisine": biz["cuisine"],
+                    "price_range": biz["price_range"],
+                    "rank_index": None,
+                    "score": None,
+                    "notes": biz["notes"],
+                    "tags": biz["tags"],
+                },
+            )
+            upserted += 1
+
+        profile.last_refreshed_at = datetime.now(timezone.utc)
+        profile.last_error = None
+        await profile.save()
+
+        logger.info(
+            "Beli profile %s refreshed: %d restaurants upserted", profile_id, upserted
+        )
+        return {"ok": True, "upserted": upserted}
+
+    except Exception as e:
+        profile.last_error = str(e)
+        await profile.save()
+        logger.error("Failed to refresh Beli profile %s: %s", profile_id, e)
+        return {"error": str(e)}
+
+
+async def _refresh_after_delay(profile_id: str, delay: float) -> None:
+    await asyncio.sleep(delay)
+    profile = await BeliProfile.get_or_none(id=profile_id)
+    if not profile:
+        return
+    await refresh_profile(profile_id)
+    # Friend sampling only runs during nightly scheduled refreshes, not manual ones
+    try:
+        access_token = await _get_access_token(profile)
+        await _refresh_friend_sample(profile, access_token)
+    except Exception as e:
+        logger.warning("Friend sample failed for profile %s: %s", profile_id, e)
+
+
+async def schedule_nightly_refresh() -> dict:
+    """Pick the NIGHTLY_BATCH_SIZE stalest profiles and schedule them with random delays."""
+    profiles = (
+        await BeliProfile.filter(enabled=True)
+        .order_by("last_refreshed_at")
+        .limit(NIGHTLY_BATCH_SIZE)
+        .all()
+    )
+
+    if not profiles:
+        logger.info("No enabled Beli profiles to refresh")
+        return {"scheduled": 0}
+
+    for profile in profiles:
+        delay = random.uniform(0, NIGHTLY_WINDOW_SECONDS)
+        asyncio.create_task(_refresh_after_delay(str(profile.id), delay))
+        logger.info("Scheduled Beli refresh for profile %s in %.0fs", profile.id, delay)
+
+    return {"scheduled": len(profiles)}

blueprints/beli/models.py

@@ -0,0 +1,66 @@
+import enum
+import uuid
+
+from tortoise import fields
+from tortoise.models import Model
+
+from blueprints.email.crypto_service import EncryptedTextField
+
+
+class BeliStatus(str, enum.Enum):
+    BEEN = "been"
+    WANT_TO_TRY = "want_to_try"
+
+
+class BeliProfile(Model):
+    id = fields.UUIDField(primary_key=True, default=uuid.uuid4)
+    user = fields.ForeignKeyField("models.User", related_name="beli_profiles")
+    beli_email = fields.CharField(max_length=255)
+    beli_password = EncryptedTextField()
+    beli_user_id = fields.CharField(max_length=255, null=True)
+    # refresh token lasts 7 days; we store it encrypted to avoid re-authenticating each run
+    beli_refresh_token = EncryptedTextField(null=True)
+    last_refreshed_at = fields.DatetimeField(null=True)
+    last_error = fields.TextField(null=True)
+    enabled = fields.BooleanField(default=True)
+    created_at = fields.DatetimeField(auto_now_add=True)
+    updated_at = fields.DatetimeField(auto_now=True)
+
+    class Meta:
+        table = "beli_profiles"
+        unique_together = (("user", "beli_email"),)
+
+
+class BeliRestaurant(Model):
+    id = fields.UUIDField(primary_key=True, default=uuid.uuid4)
+    profile = fields.ForeignKeyField("models.BeliProfile", related_name="restaurants")
+    beli_business_id = fields.CharField(max_length=255)
+    # Empty string = profile owner's own ranking; UUID string = a followed user's ranking
+    friend_beli_user_id = fields.CharField(max_length=255, default="")
+    friend_name = fields.CharField(max_length=255, null=True)
+    # Beli category code: RES / DES / BAR / BAK; empty for bookmarks
+    beli_category = fields.CharField(max_length=10, default="")
+    name = fields.CharField(max_length=500)
+    city = fields.CharField(max_length=255, null=True)
+    neighborhood = fields.CharField(max_length=255, null=True)
+    cuisine = fields.JSONField(default=list)
+    price_range = fields.CharField(max_length=10, null=True)
+    status = fields.CharEnumField(enum_type=BeliStatus, max_length=20)
+    rank_index = fields.IntField(null=True)
+    score = fields.DecimalField(max_digits=5, decimal_places=2, null=True)
+    notes = fields.TextField(null=True)
+    tags = fields.JSONField(default=list)
+    fetched_at = fields.DatetimeField(auto_now=True)
+    created_at = fields.DatetimeField(auto_now_add=True)
+
+    class Meta:
+        table = "beli_restaurants"
+        unique_together = (
+            (
+                "profile",
+                "beli_business_id",
+                "status",
+                "friend_beli_user_id",
+                "beli_category",
+            ),
+        )

blueprints/conversation/__init__.py

@@ -1,6 +1,9 @@
 import asyncio
 import json
 import logging
+import os
+import re
+import time
 import uuid
 
 from quart import Blueprint, jsonify, make_response, request
@@ -10,23 +13,77 @@ from quart_jwt_extended import (
 )
 
 import blueprints.users.models
-from utils.image_process import analyze_user_image
+from metrics import count_result, observe_response, time_response
+from utils.image_process import image_content_block
+from utils.strip_markdown import extract_agent_text
 from utils.image_upload import ImageValidationError, process_image
 from utils.s3_client import generate_presigned_url as s3_presigned_url
 from utils.s3_client import get_image as s3_get_image
 from utils.s3_client import upload_image as s3_upload_image
 
-from .agents import main_agent
 from .logic import (
     add_message_to_conversation,
+    get_agent_for,
+    persist_agent_tool_messages,
+    tool_result_text,
 )
-from .memory import get_memories_for_user
+from .memory import build_system_prompt_with_memories
 from .models import (
     Conversation,
     PydConversation,
     PydListConversation,
+    Speaker,
 )
-from .prompts import SIMBA_SYSTEM_PROMPT
+
+TTFT_TIMEOUT = float(os.getenv("TTFT_TIMEOUT_SECONDS", "30"))
+
+_THINK_OPEN = "<think>"
+_THINK_CLOSE = "</think>"
+_THINK_RE = re.compile(r"<think>.*?</think>", re.DOTALL)
+
+
+def _route_content(
+    buf: str, text: str, in_thinking: bool
+) -> tuple[str, str, bool, str]:
+    """Split a streaming content chunk into (thinking_text, content_text).
+
+    Handles tags split across chunk boundaries by keeping a small trailing
+    buffer. Returns the updated in_thinking state and any leftover buffer.
+    """
+    combined = buf + text
+    thinking_parts: list[str] = []
+    content_parts: list[str] = []
+
+    while combined:
+        tag, hold = (
+            (_THINK_CLOSE, len(_THINK_CLOSE) - 1)
+            if in_thinking
+            else (_THINK_OPEN, len(_THINK_OPEN) - 1)
+        )
+        idx = combined.find(tag)
+        if idx == -1:
+            # Tag not yet complete — hold back enough bytes to catch a split.
+            if len(combined) <= hold:
+                return (
+                    "".join(thinking_parts),
+                    "".join(content_parts),
+                    in_thinking,
+                    combined,
+                )
+            safe, leftover = combined[:-hold], combined[-hold:]
+            (thinking_parts if in_thinking else content_parts).append(safe)
+            return (
+                "".join(thinking_parts),
+                "".join(content_parts),
+                in_thinking,
+                leftover,
+            )
+        (thinking_parts if in_thinking else content_parts).append(combined[:idx])
+        combined = combined[idx + len(tag) :]
+        in_thinking = not in_thinking
+
+    return "".join(thinking_parts), "".join(content_parts), in_thinking, ""
+
 
 conversation_blueprint = Blueprint(
     "conversation_api", __name__, url_prefix="/api/conversation"
@@ -44,51 +101,28 @@ def _persist_in_background(coro) -> asyncio.Task:
     return task
 
 
-_SYSTEM_PROMPT = SIMBA_SYSTEM_PROMPT
-
-
 async def _get_owned_conversation(conversation_id, user_id) -> Conversation | None:
     """Fetch a conversation only if it belongs to the given user."""
     return await Conversation.get_or_none(id=conversation_id, user_id=user_id)
 
 
-async def _build_system_prompt_with_memories(user_id: str) -> str:
-    """Append user memories to the base system prompt."""
-    memories = await get_memories_for_user(user_id)
-    if not memories:
-        return _SYSTEM_PROMPT
-    memory_block = "\n".join(f"- {m}" for m in memories)
-    return f"{_SYSTEM_PROMPT}\n\nUSER MEMORIES (facts the user has asked you to remember):\n{memory_block}"
-
-
-def _build_messages_payload(
-    conversation,
+def _build_user_message(
     query_text: str,
-    image_description: str | None = None,
-    system_prompt: str | None = None,
-) -> list:
-    recent_messages = (
-        conversation.messages[-10:]
-        if len(conversation.messages) > 10
-        else conversation.messages
-    )
-    messages_payload = [{"role": "system", "content": system_prompt or _SYSTEM_PROMPT}]
-    for msg in recent_messages[:-1]:  # Exclude the message we just added
-        role = "user" if msg.speaker == "user" else "assistant"
-        text = msg.text
-        if msg.image_key and role == "user":
-            text = f"[User sent an image]\n{text}"
-        messages_payload.append({"role": role, "content": text})
+    image_block: dict | None = None,
+) -> dict:
+    """Build the single new user message for this turn.
 
-    # Build the current user message with optional image description
-    if image_description:
-        content = f"[Image analysis: {image_description}]"
-        if query_text:
-            content = f"{query_text}\n\n{content}"
+    History is restored by the LangGraph checkpointer; only the current
+    message is passed to avoid duplicating prior turns in checkpoint state.
+    """
+    if image_block:
+        content = [
+            {"type": "text", "text": query_text or "The user sent this image."},
+            image_block,
+        ]
     else:
         content = query_text
-    messages_payload.append({"role": "user", "content": content})
-    return messages_payload
+    return {"role": "user", "content": content}
 
 
 @conversation_blueprint.post("/query")
@@ -102,7 +136,6 @@ async def query():
     conversation = await _get_owned_conversation(conversation_id, current_user_uuid)
     if conversation is None:
         return jsonify({"error": "conversation not found"}), 404
-    await conversation.fetch_related("messages")
     await add_message_to_conversation(
         conversation=conversation,
         message=query,
@@ -110,24 +143,30 @@ async def query():
         user=user,
     )
 
-    system_prompt = await _build_system_prompt_with_memories(str(user.id))
-    messages_payload = _build_messages_payload(
-        conversation, query, system_prompt=system_prompt
-    )
-    payload = {"messages": messages_payload}
-    agent_config = {
-        "configurable": {"user_id": str(user.id), "is_admin": user.is_admin()}
-    }
+    with time_response(stream=False):
+        system_prompt = await build_system_prompt_with_memories(str(user.id))
+        payload = {"messages": [_build_user_message(query)]}
+        agent_config = {
+            "configurable": {
+                "user_id": str(user.id),
+                "is_admin": user.is_admin(),
+                "thread_id": str(conversation.id),
+                "system_prompt": system_prompt,
+            }
+        }
 
-    response = await main_agent.ainvoke(payload, config=agent_config)
-    message = response.get("messages", [])[-1].content
-    await add_message_to_conversation(
-        conversation=conversation,
-        message=message,
-        speaker="simba",
-        user=user,
-    )
-    return jsonify({"response": message})
+        response = await get_agent_for(user).ainvoke(payload, config=agent_config)
+        message = extract_agent_text(response.get("messages", [])[-1].content)
+        await persist_agent_tool_messages(
+            conversation, user, response.get("messages", [])
+        )
+        await add_message_to_conversation(
+            conversation=conversation,
+            message=message,
+            speaker="simba",
+            user=user,
+        )
+        return jsonify({"response": message})
 
 
 @conversation_blueprint.post("/upload-image")
@@ -181,7 +220,6 @@ async def stream_query():
     conversation = await _get_owned_conversation(conversation_id, current_user_uuid)
     if conversation is None:
         return jsonify({"error": "conversation not found"}), 404
-    await conversation.fetch_related("messages")
     await add_message_to_conversation(
         conversation=conversation,
         message=query_text or "",
@@ -190,29 +228,50 @@ async def stream_query():
         image_key=image_key,
     )
 
-    # If an image was uploaded, analyze it with the vision model
-    image_description = None
+    # If an image was uploaded, attach it to the agent request directly
+    image_block = None
     if image_key:
         try:
             image_bytes, _ = await s3_get_image(image_key)
-            image_description = await analyze_user_image(image_bytes)
-            logging.info(f"Image analysis complete for {image_key}")
+            image_block = image_content_block(image_bytes)
         except Exception as e:
-            logging.error(f"Failed to analyze image: {e}")
-            image_description = "[Image could not be analyzed]"
+            logging.error(f"Failed to load image {image_key}: {e}")
+            query_text = f"{query_text or ''}\n\n[An image was sent but could not be loaded.]".strip()
 
-    system_prompt = await _build_system_prompt_with_memories(str(user.id))
-    messages_payload = _build_messages_payload(
-        conversation, query_text or "", image_description, system_prompt=system_prompt
-    )
-    payload = {"messages": messages_payload}
+    system_prompt = await build_system_prompt_with_memories(str(user.id))
+    payload = {"messages": [_build_user_message(query_text or "", image_block)]}
     agent_config = {
-        "configurable": {"user_id": str(user.id), "is_admin": user.is_admin()}
+        "configurable": {
+            "user_id": str(user.id),
+            "is_admin": user.is_admin(),
+            "thread_id": str(conversation.id),
+            "system_prompt": system_prompt,
+        }
     }
+    # Capture the agent reference while the request/app context is still active.
+    # Quart exits the app context before iterating the SSE response body, so
+    # current_app cannot be accessed inside event_generator().
+    main_agent = get_agent_for(user)
 
     async def event_generator():
         final_message = None
         streamed = ""
+        # Tag-routing state for <think>…</think> blocks.
+        in_thinking = False
+        tag_buf = ""
+        # Fallback heartbeat for models that emit empty chunks before thinking text.
+        last_thinking = 0.0
+        turn_has_content = False
+
+        # Timing / observability
+        t0 = time.monotonic()
+        llm_start_at: float | None = None
+        first_content_at: float | None = None
+        tool_timers: dict[str, float] = {}  # run_id -> start time
+        tool_spans: list[dict] = []
+        n_tokens = 0
+        model_name: str | None = None
+        cancelled = False
 
         async def persist(text: str) -> None:
             await add_message_to_conversation(
@@ -222,24 +281,122 @@ async def stream_query():
                 user=user,
             )
 
+        def _log_summary(*, error: str | None = None) -> None:
+            now = time.monotonic()
+            thinking_ms = (
+                round((first_content_at - llm_start_at) * 1000)
+                if llm_start_at and first_content_at
+                else None
+            )
+            ttft_ms = (
+                round((first_content_at - t0) * 1000) if first_content_at else None
+            )
+            logging.info(
+                json.dumps(
+                    {
+                        "event": "stream_request",
+                        "conv": str(conversation.id),
+                        "user": user.username,
+                        "total_ms": round((now - t0) * 1000),
+                        "thinking_ms": thinking_ms,
+                        "ttft_ms": ttft_ms,
+                        "tokens": n_tokens,
+                        "tools": tool_spans,
+                        "model": model_name,
+                        "cancelled": cancelled,
+                        **({"error": error} if error else {}),
+                    }
+                )
+            )
+
         try:
-            async for event in main_agent.astream_events(
+            aiter = main_agent.astream_events(
                 payload, version="v2", config=agent_config
-            ):
+            ).__aiter__()
+            while True:
+                # Apply a short TTFT timeout only during the window between
+                # on_chat_model_start and the first content token. Before the
+                # LLM call starts (tool phase) and after the first token, let
+                # the httpx read timeout handle stalls instead.
+                if llm_start_at is not None and first_content_at is None:
+                    ttft_wait = TTFT_TIMEOUT - (time.monotonic() - llm_start_at)
+                    timeout = max(ttft_wait, 1.0)
+                else:
+                    timeout = None
+
+                try:
+                    if timeout is not None:
+                        event = await asyncio.wait_for(
+                            aiter.__anext__(), timeout=timeout
+                        )
+                    else:
+                        event = await aiter.__anext__()
+                except StopAsyncIteration:
+                    break
+                except asyncio.TimeoutError:
+                    _log_summary(error=f"TTFT timeout after {TTFT_TIMEOUT:.0f}s")
+                    yield f"data: {json.dumps({'type': 'error', 'message': 'Simba took too long to respond. Please try again.', 'stalled': True})}\n\n"
+                    return
+
                 event_type = event.get("event")
 
                 if event_type == "on_tool_start":
+                    tool_timers[event["run_id"]] = time.monotonic()
                     yield f"data: {json.dumps({'type': 'tool_start', 'tool': event['name']})}\n\n"
 
                 elif event_type == "on_tool_end":
+                    start = tool_timers.pop(event["run_id"], None)
+                    tool_spans.append(
+                        {
+                            "tool": event["name"],
+                            "ms": round((time.monotonic() - start) * 1000)
+                            if start
+                            else None,
+                        }
+                    )
                     yield f"data: {json.dumps({'type': 'tool_end', 'tool': event['name']})}\n\n"
+                    # Persist tool output so later turns keep the fetched context
+                    output = event.get("data", {}).get("output")
+                    if output is not None:
+                        await add_message_to_conversation(
+                            conversation=conversation,
+                            message=tool_result_text(event["name"], output),
+                            speaker="tool",
+                            user=user,
+                        )
+
+                elif event_type == "on_chat_model_start":
+                    turn_has_content = False
+                    if llm_start_at is None:
+                        llm_start_at = time.monotonic()
+                    if model_name is None:
+                        model_name = event.get("metadata", {}).get(
+                            "ls_model_name"
+                        ) or event.get("name")
 
                 elif event_type == "on_chat_model_stream":
                     chunk = event.get("data", {}).get("chunk")
-                    content = getattr(chunk, "content", None)
-                    if isinstance(content, str) and content:
-                        streamed += content
-                        yield f"data: {json.dumps({'type': 'content', 'delta': content})}\n\n"
+                    raw = getattr(chunk, "content", None)
+                    if isinstance(raw, str) and raw:
+                        think_text, content_text, in_thinking, tag_buf = _route_content(
+                            tag_buf, raw, in_thinking
+                        )
+                        if think_text:
+                            yield f"data: {json.dumps({'type': 'thinking_delta', 'delta': think_text})}\n\n"
+                        if content_text:
+                            if first_content_at is None:
+                                first_content_at = time.monotonic()
+                            turn_has_content = True
+                            n_tokens += 1
+                            streamed += content_text
+                            yield f"data: {json.dumps({'type': 'content', 'delta': content_text})}\n\n"
+                    elif not turn_has_content:
+                        # Empty chunk before any content — heartbeat for models
+                        # that don't emit <think> tags at all.
+                        now = time.monotonic()
+                        if now - last_thinking >= 1.0:
+                            last_thinking = now
+                            yield f"data: {json.dumps({'type': 'thinking'})}\n\n"
 
                 elif event_type == "on_chain_end":
                     output = event.get("data", {}).get("output")
@@ -249,19 +406,20 @@ async def stream_query():
                             last_msg = msgs[-1]
                             content = getattr(last_msg, "content", None)
                             if isinstance(content, str) and content:
-                                final_message = content
+                                final_message = _THINK_RE.sub("", content).strip()
 
         except asyncio.CancelledError:
-            # The client disconnected mid-stream (e.g. a route change aborts the
-            # fetch). CancelledError is not an Exception, so without this branch
-            # persistence below would be skipped and the answer lost. Save what
-            # was generated in a detached task, then honor the cancellation.
+            cancelled = True
+            _log_summary()
             text = final_message or streamed
             if text:
                 _persist_in_background(persist(text))
             raise
         except Exception as e:
+            count_result(stream=True, result="error")
+            _log_summary(error=str(e))
             yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"
+            return
 
         final_message = final_message or streamed
         if final_message:
@@ -273,6 +431,9 @@ async def stream_query():
         else:
             yield f"data: {json.dumps({'type': 'error', 'message': 'No response generated'})}\n\n"
 
+        observe_response(stream=True, seconds=time.monotonic() - t0)
+        count_result(stream=True, result="success")
+        _log_summary()
         yield "data: [DONE]\n\n"
 
     return await make_response(
@@ -298,10 +459,13 @@ async def get_conversation(conversation_id: str):
     # Manually serialize the conversation with messages
     messages = []
     for msg in conversation.messages:
+        # Tool messages hold raw fetched context for the LLM; only surface
+        # their header line (e.g. "[web_search result]") in the UI.
+        text = msg.text.split("\n", 1)[0] if msg.speaker == Speaker.TOOL else msg.text
         messages.append(
             {
                 "id": str(msg.id),
-                "text": msg.text,
+                "text": text,
                 "speaker": msg.speaker.value,
                 "created_at": msg.created_at.isoformat(),
                 "image_key": msg.image_key,

blueprints/conversation/logic.py

@@ -1,11 +1,55 @@
 from datetime import datetime, timezone
 
 import tortoise.exceptions
+from langchain_core.messages import ToolMessage
+from quart import current_app
 
 import blueprints.users.models
 
 from .models import Conversation, ConversationMessage
 
+# Cap persisted tool output so replayed history stays within the LLM context
+# budget (llama-server slots are 32k tokens; base prompt + tools already use ~6k).
+TOOL_RESULT_MAX_CHARS = 4000
+
+
+def tool_result_text(tool_name: str, output) -> str:
+    """Format a tool's output for persistence as a conversation message."""
+    content = getattr(output, "content", output)
+    if isinstance(content, (list, tuple)):
+        content = "\n".join(str(item) for item in content)
+    text = str(content)
+    if len(text) > TOOL_RESULT_MAX_CHARS:
+        text = text[:TOOL_RESULT_MAX_CHARS] + "\n[truncated]"
+    return f"[{tool_name} result]\n{text}"
+
+
+def get_agent_for(user: blueprints.users.models.User):
+    """Admins get the agent with admin-only tools (Homebox); everyone else the main one."""
+    if user.is_admin():
+        return getattr(current_app, "admin_agent", current_app.main_agent)
+    return current_app.main_agent
+
+
+async def persist_agent_tool_messages(
+    conversation: Conversation,
+    user: blueprints.users.models.User,
+    response_messages: list,
+) -> None:
+    """Persist ToolMessages from an agent response as 'tool' conversation messages.
+
+    Keeps fetched context (web searches, RAG chunks, etc.) available to later
+    turns, which otherwise only see the final text answers.
+    """
+    for msg in response_messages:
+        if isinstance(msg, ToolMessage):
+            await add_message_to_conversation(
+                conversation=conversation,
+                message=tool_result_text(msg.name or "tool", msg),
+                speaker="tool",
+                user=user,
+            )
+
 
 async def create_conversation(name: str = "") -> Conversation:
     conversation = await Conversation.create(name=name)
@@ -18,9 +62,8 @@ async def add_message_to_conversation(
     speaker: str,
     user: blueprints.users.models.User,
     image_key: str | None = None,
+    message_handle: str | None = None,
 ) -> ConversationMessage:
-    print(conversation, message, speaker)
-
     # Bump updated_at on every message so conversations sort by last activity.
     # `update_fields` bypasses the auto_now recompute, so set it explicitly.
     update_fields = ["updated_at"]
@@ -38,6 +81,7 @@ async def add_message_to_conversation(
         speaker=speaker,
         conversation=conversation,
         image_key=image_key,
+        message_handle=message_handle,
     )
 
     return message
@@ -71,7 +115,9 @@ async def get_conversation_for_user(user: blueprints.users.models.User) -> Conve
 async def get_conversation_for_channel(
     user: blueprints.users.models.User, channel: str
 ) -> Conversation:
-    conversation = await Conversation.filter(user=user, channel=channel).first()
+    conversation = await (
+        Conversation.filter(user=user, channel=channel).order_by("-updated_at").first()
+    )
     if conversation is None:
         conversation = await Conversation.create(
             name=f"{user.username}'s {channel} chat", user=user, channel=channel

blueprints/conversation/memory.py

@@ -1,4 +1,5 @@
 from .models import UserMemory
+from .prompts import SIMBA_SYSTEM_PROMPT
 
 
 async def get_memories_for_user(user_id: str) -> list[str]:
@@ -7,6 +8,14 @@ async def get_memories_for_user(user_id: str) -> list[str]:
     return [m.content for m in memories]
 
 
+async def build_system_prompt_with_memories(user_id: str) -> str:
+    memories = await get_memories_for_user(user_id)
+    if not memories:
+        return SIMBA_SYSTEM_PROMPT
+    block = "\n".join(f"- {m}" for m in memories)
+    return f"{SIMBA_SYSTEM_PROMPT}\n\nUSER MEMORIES (facts the user has asked you to remember):\n{block}"
+
+
 async def save_memory(user_id: str, content: str) -> str:
     """Save a new memory or touch an existing one (exact-match dedup)."""
     existing = await UserMemory.filter(user_id=user_id, content=content).first()

blueprints/conversation/models.py

@@ -11,6 +11,7 @@ from tortoise.models import Model
 class Speaker(enum.Enum):
     USER = "user"
     SIMBA = "simba"
+    TOOL = "tool"
 
 
 class Conversation(Model):
@@ -36,6 +37,7 @@ class ConversationMessage(Model):
     created_at = fields.DatetimeField(auto_now_add=True)
     speaker = fields.CharEnumField(enum_type=Speaker, max_length=10)
     image_key = fields.CharField(max_length=512, null=True, default=None)
+    message_handle = fields.CharField(max_length=255, null=True, default=None)
 
     class Meta:
         table = "conversation_messages"

blueprints/conversation/prompts.py

@@ -49,6 +49,9 @@ You have access to Ryan's meal plans, recipes, and shopping lists through Mealie
 - Use mealie_create_meal_plan to add meals to the plan
 Always use these tools when asked about meals, recipes, what's for dinner, or shopping lists.
 
+HOME INVENTORY (Homebox Integration):
+If homebox_* tools are available, use them for questions about household items: where something is stored, what's in a location, warranties, and maintenance. Search before creating items to avoid duplicates.
+
 NOTES & RESEARCH (Obsidian Integration):
 You have access to Ryan's Obsidian vault through the Obsidian integration. When users ask about research, personal notes, or information that might be stored in markdown files, use the appropriate Obsidian tools:
 - Use obsidian_search_notes to search through your vault for relevant information

blueprints/cron/__init__.py

@@ -0,0 +1,176 @@
+import hmac
+import logging
+import os
+from datetime import datetime, timezone
+
+from quart import Blueprint, current_app, jsonify, request
+
+from blueprints.users.decorators import admin_required
+from blueprints.scheduled_messages.models import MessageChannel
+from .models import CronJob
+
+cron_blueprint = Blueprint("cron_api", __name__, url_prefix="/api/cron")
+
+logger = logging.getLogger(__name__)
+
+
+def _serialize(job: CronJob) -> dict:
+    return {
+        "id": str(job.id),
+        "name": job.name,
+        "description": job.description,
+        "prompt": job.prompt,
+        "recipient": job.recipient,
+        "channel": job.channel.value,
+        "enabled": job.enabled,
+        "last_run_at": job.last_run_at.isoformat() if job.last_run_at else None,
+        "last_status": job.last_status,
+        "last_error": job.last_error,
+        "created_at": job.created_at.isoformat(),
+        "updated_at": job.updated_at.isoformat(),
+    }
+
+
+def _check_cron_secret() -> bool:
+    expected = os.environ.get("CRON_SECRET", "")
+    if not expected:
+        return False
+    provided = request.headers.get("X-Cron-Secret", "")
+    return hmac.compare_digest(provided.encode(), expected.encode())
+
+
+@cron_blueprint.route("/", methods=["GET"])
+@admin_required
+async def list_jobs():
+    jobs = await CronJob.all().order_by("name")
+    return jsonify([_serialize(j) for j in jobs])
+
+
+@cron_blueprint.route("/", methods=["POST"])
+@admin_required
+async def create_job():
+    data = await request.get_json()
+    if not data:
+        return jsonify({"error": "Invalid payload"}), 400
+
+    name = (data.get("name") or "").strip()
+    prompt = (data.get("prompt") or "").strip()
+    recipient = (data.get("recipient") or "").strip()
+    channel_str = data.get("channel")
+    description = (data.get("description") or "").strip() or None
+
+    if not name or not prompt or not recipient or not channel_str:
+        return jsonify(
+            {"error": "name, prompt, recipient, and channel are required"}
+        ), 400
+
+    try:
+        channel = MessageChannel(channel_str)
+    except ValueError:
+        return jsonify(
+            {"error": f"Invalid channel: {channel_str}. Must be 'imessage' or 'email'"}
+        ), 400
+
+    job = await CronJob.create(
+        name=name,
+        description=description,
+        prompt=prompt,
+        recipient=recipient,
+        channel=channel,
+        enabled=bool(data.get("enabled", True)),
+    )
+    return jsonify(_serialize(job)), 201
+
+
+@cron_blueprint.route("/<job_id>", methods=["PUT"])
+@admin_required
+async def update_job(job_id: str):
+    job = await CronJob.get_or_none(id=job_id)
+    if not job:
+        return jsonify({"error": "Not found"}), 404
+
+    data = await request.get_json()
+    if not data:
+        return jsonify({"error": "Invalid payload"}), 400
+
+    if "name" in data:
+        job.name = data["name"].strip()
+    if "description" in data:
+        job.description = (data["description"] or "").strip() or None
+    if "prompt" in data:
+        job.prompt = data["prompt"].strip()
+    if "recipient" in data:
+        job.recipient = data["recipient"].strip()
+    if "channel" in data:
+        try:
+            job.channel = MessageChannel(data["channel"])
+        except ValueError:
+            return jsonify({"error": f"Invalid channel: {data['channel']}"}), 400
+    if "enabled" in data:
+        job.enabled = bool(data["enabled"])
+
+    await job.save()
+    return jsonify(_serialize(job))
+
+
+@cron_blueprint.route("/<job_id>", methods=["DELETE"])
+@admin_required
+async def delete_job(job_id: str):
+    job = await CronJob.get_or_none(id=job_id)
+    if not job:
+        return jsonify({"error": "Not found"}), 404
+
+    await job.delete()
+    return jsonify({"status": "deleted"})
+
+
+@cron_blueprint.route("/<job_id>/trigger", methods=["POST"])
+async def trigger_job(job_id: str):
+    if not _check_cron_secret():
+        return jsonify({"error": "Unauthorized"}), 401
+
+    job = await CronJob.get_or_none(id=job_id)
+    if not job:
+        return jsonify({"error": "Not found"}), 404
+
+    if not job.enabled:
+        return jsonify({"skipped": True}), 200
+
+    try:
+        import uuid
+
+        response = await current_app.main_agent.ainvoke(
+            {"messages": [{"role": "user", "content": job.prompt}]},
+            config={"configurable": {"thread_id": f"cron-{uuid.uuid4()}"}},
+        )
+        result = response.get("messages", [])[-1].content
+
+        if job.channel == MessageChannel.IMESSAGE:
+            from blueprints.imessage import send_imessage
+            from utils.strip_markdown import strip_markdown
+
+            await send_imessage(job.recipient, strip_markdown(result))
+
+        elif job.channel == MessageChannel.EMAIL:
+            from blueprints.email import send_email_reply
+
+            await send_email_reply(
+                to=job.recipient,
+                subject=job.name,
+                body=result,
+            )
+
+        job.last_run_at = datetime.now(timezone.utc)
+        job.last_status = "success"
+        job.last_error = None
+        await job.save()
+        logger.info(f"Cron job {job.id} ({job.name}) executed successfully")
+        return jsonify({"ok": True})
+
+    except Exception as e:
+        job.last_run_at = datetime.now(timezone.utc)
+        job.last_status = "failed"
+        job.last_error = str(e)
+        await job.save()
+        logger.error(f"Cron job {job.id} ({job.name}) failed: {e}")
+        return jsonify({"error": str(e)}), 500

blueprints/cron/models.py

@@ -0,0 +1,22 @@
+from tortoise import fields
+from tortoise.models import Model
+
+from blueprints.scheduled_messages.models import MessageChannel
+
+
+class CronJob(Model):
+    id = fields.UUIDField(primary_key=True)
+    name = fields.CharField(max_length=255)
+    description = fields.TextField(null=True)
+    prompt = fields.TextField()
+    recipient = fields.CharField(max_length=255)
+    channel = fields.CharEnumField(enum_type=MessageChannel, max_length=20)
+    enabled = fields.BooleanField(default=True)
+    last_run_at = fields.DatetimeField(null=True)
+    last_status = fields.CharField(max_length=50, null=True)
+    last_error = fields.TextField(null=True)
+    created_at = fields.DatetimeField(auto_now_add=True)
+    updated_at = fields.DatetimeField(auto_now=True)
+
+    class Meta:
+        table = "cron_jobs"

blueprints/email/__init__.py

@@ -7,14 +7,15 @@ import time
 from collections import defaultdict
 
 import httpx
-from quart import Blueprint, request
+from quart import Blueprint, current_app, request
 
 from blueprints.users.models import User
 from blueprints.conversation.logic import (
     get_conversation_for_channel,
     add_message_to_conversation,
+    persist_agent_tool_messages,
 )
-from blueprints.conversation.agents import main_agent
+from blueprints.conversation.models import Speaker
 from blueprints.conversation.prompts import SIMBA_SYSTEM_PROMPT
 from . import models  # noqa: F401 — register Tortoise ORM models
 from .helpers import generate_email_token, get_user_email_address  # noqa: F401
@@ -193,17 +194,20 @@ async def webhook():
     # Build messages payload
     try:
         messages = await conversation.messages.all()
-        recent_messages = list(messages)[-10:]
+        recent_messages = list(messages)[-20:]
 
         messages_payload = [{"role": "system", "content": SIMBA_SYSTEM_PROMPT}]
         for msg in recent_messages[:-1]:
-            role = "user" if msg.speaker == "user" else "assistant"
+            role = "user" if msg.speaker == Speaker.USER else "assistant"
             messages_payload.append({"role": role, "content": msg.text})
         messages_payload.append({"role": "user", "content": body})
 
         logger.info(f"Invoking LangChain agent with {len(messages_payload)} messages")
-        response = await main_agent.ainvoke({"messages": messages_payload})
+        response = await current_app.main_agent.ainvoke({"messages": messages_payload})
         response_text = response.get("messages", [])[-1].content
+        await persist_agent_tool_messages(
+            conversation, user, response.get("messages", [])
+        )
     except Exception as e:
         logger.error(f"Error invoking agent for email: {e}")
         response_text = "Sorry, I'm having trouble thinking right now."

blueprints/email/crypto_service.py

@@ -4,11 +4,27 @@ Encryption service for email credentials.
 Provides transparent Fernet encryption for sensitive fields in the database.
 """
 
+import base64
 import os
 from cryptography.fernet import Fernet
 from tortoise import fields
 
 
+def _is_fernet_token(value: str) -> bool:
+    """Return True if value looks like a Fernet token.
+
+    Fernet tokens are URL-safe base64 and decode to at least 73 bytes
+    (1 version + 8 timestamp + 16 IV + 16 min-ciphertext + 32 HMAC).
+    The first decoded byte is always 0x80 (the Fernet version byte).
+    """
+    try:
+        padded = value + "=" * (-len(value) % 4)
+        raw = base64.urlsafe_b64decode(padded.encode())
+        return len(raw) >= 73 and raw[0] == 0x80
+    except Exception:
+        return False
+
+
 class EncryptedTextField(fields.TextField):
     """
     Custom Tortoise ORM field that transparently encrypts/decrypts text values.
@@ -38,10 +54,16 @@ class EncryptedTextField(fields.TextField):
         return self.fernet.encrypt(value.encode()).decode()
 
     def to_python_value(self, value: str) -> str:
-        """Decrypt value when loading from database."""
+        """Decrypt value when loading from database.
+
+        Plaintext values (e.g. freshly-set before the first DB round-trip) are
+        returned as-is. Values that look like Fernet tokens are decrypted — a
+        wrong key still raises InvalidToken, preserving the security guarantee.
+        """
         if value is None:
             return None
-        # Decrypt Fernet token
+        if not _is_fernet_token(value):
+            return value
         return self.fernet.decrypt(value.encode()).decode()
 
 

blueprints/imessage/__init__.py

@@ -1,3 +1,4 @@
+import asyncio
 import os
 import hmac
 import logging
@@ -10,11 +11,14 @@ from quart import Blueprint, request, jsonify
 
 from blueprints.users.models import User
 from blueprints.conversation.logic import (
+    get_agent_for,
     get_conversation_for_channel,
     add_message_to_conversation,
+    persist_agent_tool_messages,
 )
-from blueprints.conversation.agents import main_agent
-from blueprints.conversation.prompts import SIMBA_SYSTEM_PROMPT
+from blueprints.conversation.models import Conversation, ConversationMessage
+from blueprints.conversation.memory import build_system_prompt_with_memories
+from utils.strip_markdown import extract_agent_text, strip_markdown
 
 imessage_blueprint = Blueprint("imessage_api", __name__, url_prefix="/api/imessage")
 
@@ -57,25 +61,60 @@ def _check_rate_limit(phone_number: str) -> bool:
     return True
 
 
-async def send_imessage(to_number: str, content: str) -> dict:
+async def send_imessage(
+    to_number: str,
+    content: str,
+    reply_to_handle: str | None = None,
+) -> dict:
     """Send an iMessage via SendBlue API."""
     from_number = os.getenv("SENDBLUE_FROM_NUMBER", "")
 
+    payload: dict = {
+        "number": to_number,
+        "from_number": from_number,
+        "content": content,
+    }
+    if reply_to_handle:
+        payload["reply_to"] = {"message_handle": reply_to_handle}
+
     async with httpx.AsyncClient() as client:
         resp = await client.post(
             f"{SENDBLUE_API_BASE}/api/send-message",
             headers=_get_sendblue_headers(),
-            json={
-                "number": to_number,
-                "from_number": from_number,
-                "content": content,
-            },
+            json=payload,
             timeout=30,
         )
         resp.raise_for_status()
         return resp.json()
 
 
+async def send_typing_indicator(
+    to_number: str,
+    state: str = "start",
+    max_duration_ms: int | None = None,
+) -> None:
+    """Send or stop a typing indicator. Errors are logged and swallowed."""
+    from_number = os.getenv("SENDBLUE_FROM_NUMBER", "")
+    payload: dict = {
+        "number": to_number,
+        "from_number": from_number,
+        "state": state,
+    }
+    if max_duration_ms is not None:
+        payload["max_duration_ms"] = max_duration_ms
+
+    try:
+        async with httpx.AsyncClient() as client:
+            await client.post(
+                f"{SENDBLUE_API_BASE}/api/send-typing-indicator",
+                headers=_get_sendblue_headers(),
+                json=payload,
+                timeout=10,
+            )
+    except Exception as e:
+        logger.warning(f"Failed to send typing indicator ({state}): {e}")
+
+
 def validate_sendblue_signature(f):
     """Decorator to validate the SendBlue webhook signing secret."""
 
@@ -112,6 +151,11 @@ async def webhook():
     media_url = data.get("media_url")
     is_outbound = data.get("is_outbound", False)
 
+    # Threading fields from SendBlue
+    inbound_handle = data.get("message_handle")
+    reply_to_handle = (data.get("reply_to") or {}).get("message_handle")
+    originator_handle = (data.get("thread_originator") or {}).get("message_handle")
+
     # Ignore outbound messages (our own replies echoed back)
     if is_outbound:
         return jsonify({"status": "ignored"}), 200
@@ -181,78 +225,165 @@ async def webhook():
         await send_imessage(from_number, "Sorry, this feature is restricted to admins.")
         return jsonify({"status": "forbidden"}), 200
 
-    # Get or create conversation
+    # /threads command — list recent iMessage conversations
+    if content.strip().lower() == "/threads":
+        threads = (
+            await Conversation.filter(user=user, channel="imessage")
+            .order_by("-updated_at")
+            .limit(5)
+        )
+        if not threads:
+            await send_imessage(from_number, "No threads found.")
+        else:
+            lines = ["Your recent threads (most recent first):"]
+            for i, t in enumerate(threads, 1):
+                updated = t.updated_at.strftime("%b %-d")
+                lines.append(f"{i}. {t.name} ({updated})")
+            lines.append("\nReply to any message to continue its thread.")
+            await send_imessage(from_number, "\n".join(lines))
+        return jsonify({"status": "ok"}), 200
+
+    # Determine conversation based on iMessage reply threading:
+    # - Reply to an existing message → continue that conversation
+    # - New standalone message → start a fresh conversation
     try:
-        conversation = await get_conversation_for_channel(user=user, channel="imessage")
-        await conversation.fetch_related("messages")
+        if reply_to_handle:
+            lookup_handle = originator_handle or reply_to_handle
+            parent_msg = await ConversationMessage.get_or_none(
+                message_handle=lookup_handle,
+                conversation__user=user,
+                conversation__channel="imessage",
+            )
+            if parent_msg:
+                conversation = await Conversation.get(id=parent_msg.conversation_id)
+                logger.info(
+                    f"Continuing conversation {conversation.id} via reply to {lookup_handle!r}"
+                )
+            else:
+                # Handle not found (pre-migration messages) — fall back to most recent
+                logger.warning(
+                    f"Reply handle {lookup_handle!r} not in DB; falling back to most recent conversation"
+                )
+                conversation = await get_conversation_for_channel(
+                    user=user, channel="imessage"
+                )
+        else:
+            conversation = await Conversation.create(
+                name="New iMessage",
+                user=user,
+                channel="imessage",
+            )
+            logger.info(
+                f"New iMessage conversation {conversation.id} for {user.username}"
+            )
+
     except Exception as e:
-        logger.error(f"Failed to get conversation for user {user.username}: {e}")
+        logger.error(f"Failed to get/create conversation for {user.username}: {e}")
         await send_imessage(
             from_number, "Sorry, something went wrong. Please try again later."
         )
         return jsonify({"status": "error"}), 200
 
-    # Add user message to conversation
+    # Save the inbound message now so it's persisted before we ack
     await add_message_to_conversation(
         conversation=conversation,
         message=content,
         speaker="user",
         user=user,
+        message_handle=inbound_handle,
     )
 
-    # Analyze image if present
-    image_description = None
+    # Fire LLM processing in the background so we can ack SendBlue immediately.
+    # SendBlue retries if it doesn't receive a 200 within its timeout (~5s).
+    asyncio.ensure_future(
+        _process_and_reply(
+            from_number=from_number,
+            inbound_handle=inbound_handle,
+            conversation=conversation,
+            user=user,
+            content=content,
+            media_url=media_url,
+        )
+    )
+
+    return jsonify({"status": "ok"}), 200
+
+
+async def _process_and_reply(
+    from_number: str,
+    inbound_handle: str | None,
+    conversation: Conversation,
+    user,
+    content: str,
+    media_url: str | None,
+) -> None:
+    """Run the LLM agent and send the reply. Called as a background task."""
+    # Attach image if present
+    image_block = None
     if media_url:
         try:
-            from utils.image_process import analyze_user_image
+            from utils.image_process import image_content_block
 
             async with httpx.AsyncClient() as client:
                 resp = await client.get(media_url, timeout=30)
                 resp.raise_for_status()
-                image_bytes = resp.content
-            image_description = await analyze_user_image(image_bytes)
-            logger.info(f"Image analysis complete for iMessage from {from_number}")
+            image_block = image_content_block(resp.content)
         except Exception as e:
-            logger.error(f"Failed to analyze iMessage image: {e}")
-            image_description = "[Image could not be analyzed]"
+            logger.error(f"Failed to load iMessage image: {e}")
+            content = f"{content or ''}\n\n[An image was sent but could not be loaded.]".strip()
 
-    # Build messages payload for LangChain agent
+    await send_typing_indicator(from_number, state="start", max_duration_ms=120_000)
+
+    response_text = "Sorry, I'm having trouble thinking right now."
     try:
-        messages = await conversation.messages.all()
-        recent_messages = list(messages)[-10:]
-
-        messages_payload = [{"role": "system", "content": SIMBA_SYSTEM_PROMPT}]
-
-        for msg in recent_messages[:-1]:
-            role = "user" if msg.speaker == "user" else "assistant"
-            messages_payload.append({"role": role, "content": msg.text})
-
-        if image_description:
-            user_content = f"[Image analysis: {image_description}]"
-            if content:
-                user_content = f"{content}\n\n{user_content}"
+        if image_block:
+            user_content = [
+                {"type": "text", "text": content or "The user sent this image."},
+                image_block,
+            ]
         else:
             user_content = content
-        messages_payload.append({"role": "user", "content": user_content})
 
-        logger.info(f"Invoking LangChain agent with {len(messages_payload)} messages")
-        response = await main_agent.ainvoke({"messages": messages_payload})
-        response_text = response.get("messages", [])[-1].content
+        system_prompt = await build_system_prompt_with_memories(str(user.id))
+        system_prompt += (
+            "\n\nIMPORTANT: You are replying via iMessage. Use plain text only. "
+            "Do not use markdown tables, headers, bullet points, bold, italics, "
+            "or any other markdown formatting. Write in natural prose."
+        )
+        agent_config = {
+            "configurable": {
+                "user_id": str(user.id),
+                "is_admin": user.is_admin(),
+                "thread_id": str(conversation.id),
+                "system_prompt": system_prompt,
+            }
+        }
+        logger.info(f"Invoking LangChain agent for conversation {conversation.id}")
+        response = await get_agent_for(user).ainvoke(
+            {"messages": [{"role": "user", "content": user_content}]},
+            config=agent_config,
+        )
+        response_text = extract_agent_text(response.get("messages", [])[-1].content)
+        await persist_agent_tool_messages(
+            conversation, user, response.get("messages", [])
+        )
 
     except Exception as e:
         logger.error(f"Error invoking agent: {e}")
-        response_text = "Sorry, I'm having trouble thinking right now."
+    finally:
+        await send_typing_indicator(from_number, state="stop")
+
+    send_resp = await send_imessage(
+        from_number,
+        strip_markdown(response_text),
+        reply_to_handle=inbound_handle,
+    )
+    outbound_handle = (send_resp or {}).get("message_handle")
 
-    # Save and send response
     await add_message_to_conversation(
         conversation=conversation,
         message=response_text,
         speaker="simba",
         user=user,
+        message_handle=outbound_handle,
     )
-
-    from utils.strip_markdown import strip_markdown
-
-    await send_imessage(from_number, strip_markdown(response_text))
-
-    return jsonify({"status": "ok"}), 200

blueprints/scheduled_messages/__init__.py

@@ -1,4 +1,6 @@
+import hmac
 import logging
+import os
 from datetime import datetime, timezone
 
 from quart import Blueprint, request, jsonify
@@ -102,6 +104,11 @@ async def create_message():
         use_agent=use_agent,
         created_by_id=user_id,
     )
+
+    from utils.cronicle import create_message_event
+
+    await create_message_event(str(msg.id), msg.scheduled_at, recurrence_enum.value)
+
     return jsonify(_serialize(msg)), 201
 
 
@@ -112,13 +119,14 @@ async def update_message(msg_id: str):
     if not msg:
         return jsonify({"error": "Not found"}), 404
 
-    if msg.status != MessageStatus.PENDING:
-        return jsonify({"error": "Can only update pending messages"}), 400
+    if msg.status not in (MessageStatus.PENDING, MessageStatus.FAILED):
+        return jsonify({"error": "Can only update pending or failed messages"}), 400
 
     data = await request.get_json()
     if not data:
         return jsonify({"error": "Invalid payload"}), 400
 
+    rescheduled = False
     if "recipient" in data:
         msg.recipient = data["recipient"].strip()
     if "channel" in data:
@@ -145,18 +153,47 @@ async def update_message(msg_id: str):
             if scheduled_at <= datetime.now(timezone.utc):
                 return jsonify({"error": "scheduled_at must be in the future"}), 400
             msg.scheduled_at = scheduled_at
+            rescheduled = True
         except ValueError:
             return jsonify({"error": "Invalid scheduled_at format"}), 400
     if "status" in data and data["status"] == "cancelled":
         msg.status = MessageStatus.CANCELLED
+    elif rescheduled:
+        msg.status = MessageStatus.PENDING
+        msg.error_message = None
 
     if msg.channel == MessageChannel.EMAIL and not msg.subject:
         return jsonify({"error": "subject is required for email messages"}), 400
 
     await msg.save()
+
+    if rescheduled and msg.status == MessageStatus.PENDING:
+        from utils.cronicle import create_message_event
+
+        await create_message_event(str(msg.id), msg.scheduled_at, msg.recurrence.value)
+
     return jsonify(_serialize(msg))
 
 
+def _check_cron_secret() -> bool:
+    expected = os.environ.get("CRON_SECRET", "")
+    if not expected:
+        return False
+    provided = request.headers.get("X-Cron-Secret", "")
+    return hmac.compare_digest(provided.encode(), expected.encode())
+
+
+@scheduled_messages_blueprint.route("/trigger", methods=["POST"])
+async def trigger_due_messages():
+    if not _check_cron_secret():
+        return jsonify({"error": "Unauthorized"}), 401
+
+    from .scheduler import process_due_messages
+
+    result = await process_due_messages()
+    return jsonify(result)
+
+
 @scheduled_messages_blueprint.route("/<msg_id>", methods=["DELETE"])
 @admin_required
 async def delete_message(msg_id: str):

blueprints/scheduled_messages/scheduler.py

@@ -1,14 +1,13 @@
-import asyncio
 import logging
+import uuid
 from datetime import datetime, timezone
 from dateutil.relativedelta import relativedelta
+from quart import current_app
 
 from .models import ScheduledMessage, MessageChannel, MessageStatus, Recurrence
 
 logger = logging.getLogger(__name__)
 
-POLL_INTERVAL = 15
-
 RECURRENCE_DELTAS = {
     Recurrence.DAILY: relativedelta(days=1),
     Recurrence.WEEKLY: relativedelta(weeks=1),
@@ -16,27 +15,20 @@ RECURRENCE_DELTAS = {
 }
 
 
-async def _run_agent(prompt: str) -> str:
-    """Run a prompt through the LangChain agent and return the response text."""
-    from blueprints.conversation.agents import main_agent
-    from blueprints.conversation.prompts import SIMBA_SYSTEM_PROMPT
-
-    messages_payload = [
-        {"role": "system", "content": SIMBA_SYSTEM_PROMPT},
-        {"role": "user", "content": prompt},
-    ]
-    response = await main_agent.ainvoke({"messages": messages_payload})
+async def _run_agent(prompt: str, thread_id: str) -> str:
+    response = await current_app.main_agent.ainvoke(
+        {"messages": [{"role": "user", "content": prompt}]},
+        config={"configurable": {"thread_id": thread_id}},
+    )
     return response.get("messages", [])[-1].content
 
 
 async def _schedule_next_occurrence(msg: ScheduledMessage):
-    """Create the next pending occurrence for a recurring message."""
     delta = RECURRENCE_DELTAS.get(msg.recurrence)
     if not delta:
         return
 
     next_at = msg.scheduled_at + delta
-    # If we missed several intervals, advance until we're in the future
     now = datetime.now(timezone.utc)
     while next_at <= now:
         next_at += delta
@@ -56,58 +48,55 @@ async def _schedule_next_occurrence(msg: ScheduledMessage):
     )
 
 
-async def scheduled_messages_loop():
-    """Background loop that polls for and sends due scheduled messages."""
-    logger.info(f"Scheduled messages loop started (interval={POLL_INTERVAL}s)")
+async def process_due_messages() -> dict:
+    """Process all currently due pending messages. Returns a result summary."""
+    now = datetime.now(timezone.utc)
+    due = await ScheduledMessage.filter(
+        status=MessageStatus.PENDING,
+        scheduled_at__lte=now,
+    ).all()
 
-    while True:
+    sent, failed = 0, 0
+    for msg in due:
         try:
-            now = datetime.now(timezone.utc)
-            due = await ScheduledMessage.filter(
-                status=MessageStatus.PENDING,
-                scheduled_at__lte=now,
-            ).all()
+            send_content = msg.content
 
-            for msg in due:
-                try:
-                    send_content = msg.content
+            if msg.use_agent:
+                send_content = await _run_agent(
+                    msg.content, f"scheduled-{uuid.uuid4()}"
+                )
 
-                    if msg.use_agent:
-                        send_content = await _run_agent(msg.content)
+            if msg.channel == MessageChannel.IMESSAGE:
+                from blueprints.imessage import send_imessage
+                from utils.strip_markdown import strip_markdown
 
-                    if msg.channel == MessageChannel.IMESSAGE:
-                        from blueprints.imessage import send_imessage
-                        from utils.strip_markdown import strip_markdown
+                await send_imessage(msg.recipient, strip_markdown(send_content))
 
-                        await send_imessage(msg.recipient, strip_markdown(send_content))
+            elif msg.channel == MessageChannel.EMAIL:
+                from blueprints.email import send_email_reply
 
-                    elif msg.channel == MessageChannel.EMAIL:
-                        from blueprints.email import send_email_reply
+                await send_email_reply(
+                    to=msg.recipient,
+                    subject=msg.subject or "(no subject)",
+                    body=send_content,
+                )
 
-                        await send_email_reply(
-                            to=msg.recipient,
-                            subject=msg.subject or "(no subject)",
-                            body=send_content,
-                        )
+            msg.status = MessageStatus.SENT
+            msg.error_message = None
+            await msg.save()
+            logger.info(
+                f"Sent scheduled {msg.channel.value} message {msg.id} to {msg.recipient}"
+            )
+            sent += 1
 
-                    msg.status = MessageStatus.SENT
-                    msg.error_message = None
-                    await msg.save()
-                    logger.info(
-                        f"Sent scheduled {msg.channel.value} message {msg.id} to {msg.recipient}"
-                    )
+            if msg.recurrence != Recurrence.NONE:
+                await _schedule_next_occurrence(msg)
 
-                    # Schedule next occurrence for recurring messages
-                    if msg.recurrence != Recurrence.NONE:
-                        await _schedule_next_occurrence(msg)
+        except Exception as e:
+            msg.status = MessageStatus.FAILED
+            msg.error_message = str(e)
+            await msg.save()
+            logger.error(f"Failed to send scheduled message {msg.id}: {e}")
+            failed += 1
 
-                except Exception as e:
-                    msg.status = MessageStatus.FAILED
-                    msg.error_message = str(e)
-                    await msg.save()
-                    logger.error(f"Failed to send scheduled message {msg.id}: {e}")
-
-        except Exception:
-            logger.exception("Error in scheduled messages loop")
-
-        await asyncio.sleep(POLL_INTERVAL)
+    return {"sent": sent, "failed": failed}

blueprints/whatsapp/__init__.py

@@ -9,11 +9,12 @@ from twilio.twiml.messaging_response import MessagingResponse
 
 from blueprints.users.models import User
 from blueprints.conversation.logic import (
+    get_agent_for,
     get_conversation_for_channel,
     add_message_to_conversation,
+    persist_agent_tool_messages,
 )
-from blueprints.conversation.agents import main_agent
-from blueprints.conversation.prompts import SIMBA_SYSTEM_PROMPT
+from utils.strip_markdown import extract_agent_text
 
 whatsapp_blueprint = Blueprint("whatsapp_api", __name__, url_prefix="/api/whatsapp")
 
@@ -163,7 +164,6 @@ async def webhook():
     # Get or create a conversation for this user
     try:
         conversation = await get_conversation_for_channel(user=user, channel="whatsapp")
-        await conversation.fetch_related("messages")
     except Exception as e:
         logger.error(f"Failed to get conversation for user {user.username}: {e}")
         return _twiml_response("Sorry, something went wrong. Please try again later.")
@@ -176,27 +176,23 @@ async def webhook():
         user=user,
     )
 
-    # Build messages payload for LangChain agent with system prompt and conversation history
     try:
-        # Get last 10 messages for conversation history
-        messages = await conversation.messages.all()
-        recent_messages = list(messages)[-10:]
-
-        # Build messages payload
-        messages_payload = [{"role": "system", "content": SIMBA_SYSTEM_PROMPT}]
-
-        # Add recent conversation history (exclude the message we just added)
-        for msg in recent_messages[:-1]:
-            role = "user" if msg.speaker == "user" else "assistant"
-            messages_payload.append({"role": role, "content": msg.text})
-
-        # Add current query
-        messages_payload.append({"role": "user", "content": body})
-
-        # Invoke LangChain agent
-        logger.info(f"Invoking LangChain agent with {len(messages_payload)} messages")
-        response = await main_agent.ainvoke({"messages": messages_payload})
-        response_text = response.get("messages", [])[-1].content
+        agent_config = {
+            "configurable": {
+                "user_id": str(user.id),
+                "is_admin": user.is_admin(),
+                "thread_id": str(conversation.id),
+            }
+        }
+        logger.info(f"Invoking LangChain agent for conversation {conversation.id}")
+        response = await get_agent_for(user).ainvoke(
+            {"messages": [{"role": "user", "content": body}]},
+            config=agent_config,
+        )
+        response_text = extract_agent_text(response.get("messages", [])[-1].content)
+        await persist_agent_tool_messages(
+            conversation, user, response.get("messages", [])
+        )
 
         # Log YNAB availability
         if os.getenv("YNAB_ACCESS_TOKEN"):

config/db.py

@@ -17,6 +17,8 @@ TORTOISE_CONFIG = {
                 "blueprints.users.models",
                 "blueprints.email.models",
                 "blueprints.scheduled_messages.models",
+                "blueprints.cron.models",
+                "blueprints.beli.models",
                 "aerich.models",
             ],
             "default_connection": "default",

docker-compose.yml

@@ -46,8 +46,7 @@ x-raggr-common: &raggr-common
     - S3_REGION=${S3_REGION:-garage}
     - OLLAMA_HOST=${OLLAMA_HOST:-http://localhost:11434}
     - FERNET_KEY=${FERNET_KEY}
-    - GOOGLE_CALENDAR_ENABLED=${GOOGLE_CALENDAR_ENABLED:-}
-    - GOOGLE_WORKSPACE_CLI_CREDENTIALS_FILE=${GOOGLE_WORKSPACE_CLI_CREDENTIALS_FILE:-/app/config/gws-credentials.json}
+    - GOOGLE_TOKEN_FILE=${GOOGLE_TOKEN_FILE:-/app/config/google-token.json}
     - MAILGUN_API_KEY=${MAILGUN_API_KEY}
     - MAILGUN_DOMAIN=${MAILGUN_DOMAIN}
     - MAILGUN_WEBHOOK_SIGNING_KEY=${MAILGUN_WEBHOOK_SIGNING_KEY}
@@ -61,12 +60,18 @@ x-raggr-common: &raggr-common
     - SENDBLUE_SIGNATURE_VALIDATION=${SENDBLUE_SIGNATURE_VALIDATION:-true}
     - ALLOWED_IMESSAGE_NUMBERS=${ALLOWED_IMESSAGE_NUMBERS}
     - SIMBAKIT_URL=${SIMBAKIT_URL:-http://simbakit:8585}
+    - CRON_SECRET=${CRON_SECRET}
+    - CRONICLE_URL=${CRONICLE_URL:-http://cronicle:3012}
+    - CRONICLE_API_KEY=${CRONICLE_API_KEY}
+    - HOMEBOX_URL=${HOMEBOX_URL}
+    - HOMEBOX_TOKEN=${HOMEBOX_TOKEN}
   depends_on:
     postgres:
       condition: service_healthy
   volumes:
     - ./obvault:/app/data/obsidian
-    - ./credentials.json:/app/config/gws-credentials.json:ro
+    - ./token.json:/app/config/google-token.json:ro
+    - raggr-checkpoints:/app/checkpoints
   healthcheck:
     test: ["CMD", "curl", "-f", "http://localhost:8080/api/health"]
     interval: 5s
@@ -124,3 +129,4 @@ services:
 volumes:
   postgres_data:
   simbakit-data:
+  raggr-checkpoints:

metrics.py

@@ -0,0 +1,55 @@
+import time
+from contextlib import contextmanager
+from prometheus_client import (
+    CollectorRegistry,
+    Counter,
+    Histogram,
+    generate_latest,
+)
+
+REGISTRY = CollectorRegistry(auto_describe=True)
+
+CHAT_RESPONSE_SECONDS = Histogram(
+    "raggr_chat_response_seconds",
+    "End-to-end chat response time in seconds.",
+    labelnames=("stream",),
+    buckets=(0.5, 1, 2, 5, 10, 30, 60, 120, 300),
+    registry=REGISTRY,
+)
+
+CHAT_REQUESTS_TOTAL = Counter(
+    "raggr_chat_requests_total",
+    "Total chat requests by stream mode and result.",
+    labelnames=("stream", "result"),
+    registry=REGISTRY,
+)
+
+
+def _s(stream: bool) -> str:
+    return "true" if stream else "false"
+
+
+@contextmanager
+def time_response(stream: bool):
+    t0 = time.monotonic()
+    result = "success"
+    try:
+        yield
+    except Exception:
+        result = "error"
+        raise
+    finally:
+        CHAT_RESPONSE_SECONDS.labels(stream=_s(stream)).observe(time.monotonic() - t0)
+        CHAT_REQUESTS_TOTAL.labels(stream=_s(stream), result=result).inc()
+
+
+def observe_response(stream: bool, seconds: float) -> None:
+    CHAT_RESPONSE_SECONDS.labels(stream=_s(stream)).observe(seconds)
+
+
+def count_result(stream: bool, result: str) -> None:
+    CHAT_REQUESTS_TOTAL.labels(stream=_s(stream), result=result).inc()
+
+
+def render_metrics() -> bytes:
+    return generate_latest(REGISTRY)

migrations/models/11_20260916031522_add_message_handle_to_conversation_messages.py

@@ -0,0 +1,71 @@
+from tortoise import BaseDBAsyncClient
+
+RUN_IN_TRANSACTION = True
+
+
+async def upgrade(db: BaseDBAsyncClient) -> str:
+    return """
+        ALTER TABLE "conversation_messages" ADD "message_handle" VARCHAR(255);
+        COMMENT ON COLUMN "conversation_messages"."speaker" IS 'USER: user\nSIMBA: simba\nTOOL: tool';"""
+
+
+async def downgrade(db: BaseDBAsyncClient) -> str:
+    return """
+        ALTER TABLE "conversation_messages" DROP COLUMN "message_handle";
+        COMMENT ON COLUMN "conversation_messages"."speaker" IS 'USER: user\nSIMBA: simba';"""
+
+
+MODELS_STATE = (
+    "eJztXW1v27YW/iuEPnVAFuSlybZguICdqK1vY7uInbt1dSHQEhPrWqJUUWpqFP3vO6QlWy"
+    "+UbfktVsoPW2qShyIfkofnPDyivmuuZxGHHV979CsJGA5tj2pX6LtGsUvgH9L8I6Rh35/n"
+    "8oQQDx0hYKZKihw8ZGGAzRAyH7DDCCRZhJmB7ccPo5Hj8ETPhII2fZwnRdT+EhEj9B5JOC"
+    "IBZHz6DMk2tcg3wpKf/th4sIljZdptW/zZIt0IJ75Iu79v3bwRJfnjhobpOZFL56X9STjy"
+    "6Kx4FNnWMZfheY+EkgCHxEp1g7cy7naSNG0xJIRBRGZNteYJFnnAkcPB0P58iKjJMUDiSf"
+    "x/r/+jVYAHoObQ2jTkWHz/Me3VvM8iVeOPun7XuHt1fvmL6KXHwsdAZApEtB9CEId4Kipw"
+    "nQMp/hagvB7hQA5lUj4HJjR0HRiThDmO8zmUAJkAtB5qmou/GQ6hj+EIfp5dXCyA8X+NO4"
+    "EklBJQejCvp7O+E2edTfM4pHMIzYDwLhs4LAJ5Azmh7RI5mFnJHKRWLHqc/ONAAYY+WF3q"
+    "TOJFsADffqut9/qN9gfeE5exL46AqNHXec6ZSJ3kUl9d5oZiVgn6q9V/h/hP9E+3o+fn/q"
+    "xc/x+NtwlHoWdQ78nAVmq9JqkJMJmBjXxrzYHNSqqBfdaBjRufWrAjTClxqqi9lMhami9u"
+    "VaXh057IcIP9Iqf5TlZRfCfleu8kr/YiRgKj2j6cEtlgM14Hy82WwvpbLzdgHsbSnZeDUQ"
+    "TvjRcQ+5G+JxMBYQvagakp23Jj0+0+ruZQQZunzlsR4KeZTZeeFNA76BMJp4ux0btu3Oia"
+    "wHCIzfETDiyjBEyXMIYfCSsC2owl37y/I87MwJVjmTaD29MaD1M3l2ErsPLOvBRGGfSKWe"
+    "6Zm0/BFPptxc/mT1qAzhJfIgXiai6FkR5I5VrU2rUIyTeJ4dSHVDmUSfm6uBaLDCL9737G"
+    "Fko20lftxt+/ZOyh227nbVI8tfFe33abys94ieZo0c9gPsFjmTHA7VGdRm7BFMiMbkr8mZ"
+    "eOdt/T764Q39EHtNdqNxtXiNnuEA9ov9u9BTQ8T9jQVa3X01Ws19Ny6/W0YL3aLuwyxphM"
+    "qngBGaG9+QHbJEAuTs9WQBJKlUIp8rJYxnu2AT6S5VRik4qStUR1N7RS2iyqZpRIRLdpod"
+    "TU4TJzJPeGjleeMz9YFJd6YJLpUtUT26XfwT3cNnG9YKJJ3I1U7kIvQ7iZLi9oK+/iBXgX"
+    "8JCQ0EoORkpE+RjKx/ipfAx1lvEiBrZwlvFsLPxhKcO60PAHbAVuwMPv2vors/uWW3zK0q"
+    "u/pcfHsWqYSlqmLrbeHjgFHzP25AWSmdm0KQ4mcjDTUjkwh5Nw6k0dHEOzALwmt3vTuztA"
+    "1mx1Gncf5aZzUj49b5sf+3ojhy5xsV0pqmAmsJ0punzFb3OCnp6sRsouYmULtOzTCIcMVL"
+    "pBI3dYRonLwZSIboVM3C+q56uAel6O6bmE6Y5J1uqQSkQVpMm6NQjl3ZPpUs9zCKYL1nxa"
+    "Nq9R4yOSXexPVa2fCiq1273NqNRmK8823LebOugDATMUsqfWZavTl4I7crEJZteYSKjaJQ"
+    "o1J1vDGXt6uYpmzTudKcV6mZ+xnm2ZBouG/yemhAAoBzQvV0Mwd2JIgUs+MvzA+2pb1VRq"
+    "QXB/1qnmeCZ2NjDvc0eJq6jVi3K1elFQq46FfeMx8CJfEsf13163I4c0J5YD9J5CLz9Zth"
+    "keIcdm4eedwfvp825cJ97zxURvntPN2aq8AkX0vkg+UBG9L3RgZ9GcBaJtWShs4Q2t7cTD"
+    "HrKTX4yDzYanzA9918cie858eDN/JSim9jE2TS/iYhsBovO6GtOqagwJM0fEisAXM7YTQ9"
+    "5L6lMB5HkqX8wYTcLl6wkPVk7mi5m7GpuviepQPJzwN8ScW0aYWigOATjWciitIjOgA9oL"
+    "PdDoyMcBIxYSjUKi4JMdjtD5ya8WnqCA8PLQmGN0a9MxQ6GHrkeB5+Kb5oA+eAH6Cu4U/G"
+    "EEB+YImdjHQxs8YlBSomXq1GFLpvMGpw4J9SUDdHlAoxTYWtC6O3GX12AeNiUdnv0A52Il"
+    "fvxiAT9+UeTHH0CLcJsR5rFknypHMy9XT0h3MjfB5C4FdMHLKl4d4dx3OBl3Fqu6oYlMPR"
+    "3QmjicSbcXUglDz5oYVd/iygjVJJp/38tCYDQKXclh8RJgEyEFrDx8VVjZ8ERoWkXLTSJa"
+    "E5DVFSeKVFyXLSbffBtqW2Ngs5L1HNiaDORKm3VM6lWMBc5KqXDgBJAihpUjguvLjx7lIo"
+    "Ozk+SQgoMzGJcRi6kxWMIvZpjxVXnGWIbThA/2YxQIfhhxlq/VbnzgyZQIuouVsY4Vakhx"
+    "kDOpgFicboQ2TllIQs1g4sNEQ0ng5lGqDsRIGMLj2dGAcnYzqQamcxixY9SLfN8LQoZcmG"
+    "K275CkBEM+CcS73YqjPBCOMp601emggqBiKhPGwma+gydG1YDzvJzyHVI3LfjGCCZ2FTgz"
+    "QnVh1/YFJtfPRTBbtIQ4yMjksLR3Z5v88cf5BlCKu6V+PTt9/dvr388vX/8ORURTZim/LQ"
+    "C3GFQqMFjnRZKCoJqMuclY+kqJntgh5ZxWoYa6oLtvYssGRxvska+Subsw2jwjt8dI89nG"
+    "f8CB5g5moUGCwJME8JZP2axUTbZ5dY2A4gtVdKkaWHWNgLpG4HCvEcgRG9uIT60XltmYqQ"
+    "k1jSkrtwUgelBbb1ZZjSDZOX2cQqaMQc6Ct4xEzg3cch65D9ljhrgc8gOPKx0mIk95JUTQ"
+    "riRNFBd55Mo1cB65zWGG/xjiNu1UmO+9IOP6R9NEqMyEuog1C40FTRgzxw9QYRQQxHs3ho"
+    "bwMFc2ZY+RDb4fcQnU70wr5gLCakbirkcorljkA2GRhUcjJu064VpF6S3YdoflIR2QKbfS"
+    "UbAYkiQSOZKtk1KuUCa6P8rwZAPTfMuEYYJByWlwKYAFuZ8SPV49MSNOPBnxNiExYkpBLB"
+    "P/KbEUSzJGYX0Nna9AKelnVtKKSXmZTIqKw3phcVi7DmqofxRW4ZVXiR8tey223JGWv5Sr"
+    "7nGstZ8ZENP2bemd3eXn7xmhupwO7yMif9E3DJd/M2bT7xluD1SN77i9xlv9CiXXzQ2o3m"
+    "60bq/Q7H7E5//aobpufhfnxM/wXvCLjLGb75fV3Ym8bD0dipo4ECu5hmWHPit+Dmwmvcer"
+    "5WKrbnpIcvxB79y0Om+LFogW51whn1ALxnNAe3qnf4UYKMoBfQNaX7+5EicLxBrQ60bnWr"
+    "8VSSbvqxNfmfn82wGYJVEQkBj/dYYpW8Meh+pu9uDjTjyBc4PEkwFYj8JOfAND8vEKWTAi"
+    "kwH9S9ff859PhIz573a303/HE1zY60bO5DAGJ2LEgMko260XBstl5NS1rLlrWfnhXeKOVT"
+    "GDCoI12bZVzJwiBFXMnBrYUqY3WXbDSdXPM+YFFd+bwmQLlG/9o+gKU+SQ+N4GCWxzpElY"
+    "3jjnaBG3i+dlDobPLX93asUIiHgMn/McYTvnzuX0Lb8JVvrl1HKuJiVSFwZsH3fHw9KoAG"
+    "JcvJ4A7uQjMaVcbPkd8eVc7N7uh9/Zpru1++EL2/M+t5cf/wK3jcna"
+)

migrations/models/12_20260916223332_add_cron_jobs.py

@@ -0,0 +1,86 @@
+from tortoise import BaseDBAsyncClient
+
+RUN_IN_TRANSACTION = True
+
+
+async def upgrade(db: BaseDBAsyncClient) -> str:
+    return """
+        CREATE TABLE IF NOT EXISTS "cron_jobs" (
+    "id" UUID NOT NULL PRIMARY KEY,
+    "name" VARCHAR(255) NOT NULL,
+    "description" TEXT,
+    "prompt" TEXT NOT NULL,
+    "recipient" VARCHAR(255) NOT NULL,
+    "channel" VARCHAR(20) NOT NULL,
+    "enabled" BOOL NOT NULL DEFAULT True,
+    "last_run_at" TIMESTAMPTZ,
+    "last_status" VARCHAR(50),
+    "last_error" TEXT,
+    "created_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updated_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+COMMENT ON COLUMN "cron_jobs"."channel" IS 'IMESSAGE: imessage\nEMAIL: email';"""
+
+
+async def downgrade(db: BaseDBAsyncClient) -> str:
+    return """
+        DROP TABLE IF EXISTS "cron_jobs";"""
+
+
+MODELS_STATE = (
+    "eJztXW1v27YW/iuCPnVAFuSlybbgYoCdqK3X2C5iZ1tXFwItMbEWidQoqalR9L9fkpZsia"
+    "Jky69Syg9bapKHIh8ekec8PKS+6R62oRscX2P0BZIAhA5G+pX2TUfAg/Qf0vwjTQe+v8hl"
+    "CSEYu1zASpXkOWAchARYIc18AG4AaZINA4s4fvwwFLkuS8QWLeigx0VShJz/ImiG+BGGE0"
+    "hoxqfPNNlBNvwKg+Sn/2Q+ONC1M+12bPZsnm6GU5+n3d93bt7wkuxxY9PCbuShRWl/Gk4w"
+    "mhePIsc+ZjIs7xEiSEAI7VQ3WCvjbidJsxbThJBEcN5Ue5FgwwcQuQwM/X8PEbIYBhp/Ev"
+    "vf69/1CvBQqBm0DgoZFt++z3q16DNP1dmjrt+17l6dX/7Ee4mD8JHwTI6I/p0LghDMRDmu"
+    "CyD53xyU1xNA5FAm5QUwaUPXgTFJWOC40KEEyASg9VDTPfDVdCF6DCf059nFRQmMf7buOJ"
+    "K0FIcSU72eaX0vzjqb5TFIFxBaBLIumyDMA3lDc0LHg3Iws5ICpHYsepz8o6YA0z7YfeRO"
+    "45egBN9hp2sMhq3uB9YTLwj+czlEraHBcs546lRIfXUpDMW8Eu2vzvCdxn5q//R7hqj783"
+    "LDf3TWJhCF2ET42QR26n1NUhNgMgMb+faaA5uVVAN70IGNG596YScAIehWmfZSImvNfHGr"
+    "Kg2f/gzHG6wXwsx3ssrEd1I8752I014UQGJWW4dTIhssxutgudmrsP7SywyYhyfpysvAyI"
+    "P3BhPoPKL3cMoh7NB2AGTJltzYdLuPq6kraIvURSsIeJ7bdGmloL2jfYLh7GVsDa5bN4bO"
+    "MRwD6+kZENssANODQQAeYZAHtB1Lvnl/B925gSvHMm0Gd2c11nNuLsKWY4XPcAqjDHr5LO"
+    "/ME1MAov2242ezJ5Wgs8SXSIG4mkthpgdSuRaNdi1C+FViOA1pqhzKpHxTXIsyg8j4e5ix"
+    "hZKF9FW39fdPGXvott97mxRPLbzXt/228jNeojma9zMCH4InmTHA7FEDRV7OFMiMbkr8wK"
+    "+Ofj8w7q40tqKP0KDTbbeutMDxxmCEhv3+LUUDY25DV7VeT1exXk+LrdfTnPXqeHSVMZ/g"
+    "tIoXkBHamx+wTQLk4vRsBSRpqUIoeV4Wy3jNNqmPZLuV2KS8ZCNR3Q2tlDaLqhklEtFtWi"
+    "gNdbgsgeTe0PESOfPaorjUA5OoS1VPbJd+B/Nwu9DDZKpL3I1UbqmXwd1MjxV0lHfxArwL"
+    "+pAQokoORkpE+RjKx/ihfAy1l/EiBja3l3EwFr5ek2FTaPgaW4Eb8PC7tv6K7L7lFp+y9J"
+    "pv6bFxrBqmkpZpiq23B07BB0HwjIlEM9sOAmQqBzMtJYA5noYzb6p2DE0JeG1m96ZXdwpZ"
+    "u9Nr3X2Um85J+bTetj8OjZaALvSAUymqYC6wHRVd/sZvU0FPT1YjZctY2Rwt+zwBYUCndB"
+    "NF3riIEpeDKRHdCpm4X1TPVwH1vBjTcwnTHZOs1SGViCpIk/fWhIh1TzaXYuxCgEre+bSs"
+    "OKPGWyS7WJ+qWj8VptR+/zYzpbY7Ittw320bdD7gMNNCzsy67PSGUnAnHrCo2fUEJVTtkg"
+    "lVkG2gxp5erjKzik5namK9FDUWO7ZlBtH4X2hJCIBiQEW5BoK5E0OKuuQT0yf4i2NXm1Jz"
+    "gvuzTnUXW8DdwLwXthJXmVYviqfVi9y06trANx8JjnxJHNcfg35PDqkgJgB6j2gvP9mOFR"
+    "5prhOEn3cG76fPu3GdWM/LiV6R0xVsVVaBInpfJB+oiN4XOrDzaM4c0bYsFDZ3Qms78bB1"
+    "dvLzcbDZ8JTFpu/6WGT3meun+StBMbOPgWXhiIltBIjB6mrNqmowJIE1gXZEfTFzOzHkg6"
+    "Q+FUAuUvlcY3QJl28kPFgxmc81dzU2X+fVafFw0r8hYNyyBpCtxSEAx7qA0ioyIzRCgxDT"
+    "GV3zAQmgrfFGabzgsxNOtPOTn20w1Qhk5WljjrVbBz0FWoi16wnBHrhpj9ADJtoX6k7RPw"
+    "EExJpoFvDB2KEeMZ2keMvUrsOWTOcNdh0S6ksG6PKARimwjaB1d+Iur8E8bEo6HHwD52Il"
+    "fvyihB+/yPPjD3QWYTYj1WPJOlWMpijXTEh3opvU5C4EtOSwCm4inPsOJ2POYlU3NJFppg"
+    "PaEIcz6XYplTDG9tSseoorI9SQaP59vxYco0noSTaLlwCbCClg5eGr3MqmT6RNq2i5SUQb"
+    "ArK64kSRiuuyxfCr79Da1hjYrGQzB7YhA7nSYh2TehVjgbNSKhw4ASSPYeWI4Obyo0dCZH"
+    "BWSeoUHJzBuIhYTI3BEn4xw4yvyjPGMowmfHAeI8L5YY2xfJ1u6wNLRpDTXUER61ihhhQH"
+    "OZci0GZ0I23jjIWEyCJTnyqalgRuHqXq0AIYhvTxwdEIMXYzqYaqcxgFx9og8n1MwkDzqI"
+    "o5vguTEoHmQ8LPdiuOsiYcZay01emgnKBiKhPGwgl8F0zNqgHnopzyHVI3LfjmhCp2FTgz"
+    "Qk1h1/YFJpuf82B2UAFxkJERsHR2Z5v89tv5BlDyu6V+Pjt9/cvrX88vX/9Ki/CmzFN+KQ"
+    "E3H1TKMVjnIElOUCmjoIyFR0qMxA4p5rRyNTQF3X0TWw51tKk98kWiu6XR5hm5PUaazxf+"
+    "GgeauyAITUgIlgTwFqtsVqohy7y6RkDxhSq6VA2sukZAXSNQ32sEBGJjG/GpzcIyGzM1RZ"
+    "Y5Y+W2AMSA1jaYV9YgSHZOH6eQKWKQs+AtI5GFgVvOIw9p9lOgMTnNJ5hNOgGPPGWVQE67"
+    "wjRRnOeRK9fAeOQug5n+F2jMpp0Js7WXynj+0SyRVmbRuqA9D42lM2HMHD/QCiMCNda7J9"
+    "oQFuYazNhjzaG+H/Qgrd+dVcwEuNWs8bseaXHFIteEReYeDVfadcK18tJbsO3q5SHVyJRb"
+    "aSuYD0kSiRzJ3pNCrlAmuj/K8GQD03zLhGGCQcFucCGAObkfEj1WPbQiRjyZ8TIhMWIKQS"
+    "wS/yGx5K9kjML6M7RYgZqkDzxJKyblZTIpKg7rhcVh7TqooflRWLkjrxI/WnYsttiRlh/K"
+    "Vfc4NtrPJNByfEd6Z3fx/ntGqCm7w/uIyC/7huHyb8Zs+j3D7YGqsxV30HprXGnJdXMjZH"
+    "RbndsrbX4/4uG/dqium9/FPvEBzgW/yBi7xXpZ3Z0QZZvpUDTEgVjJNSza9Fnxc2Bz6T1e"
+    "LRdbdbNNkuMPRu+m03ubt0D0OOdK8yGy6XiO0MDoDa+0gE6UI/SGzvrGzRXfWYD2CF23et"
+    "fGLU+yWF/d+MrMwy8H1CyJCIEx/usMU7aGPQ7V3fzBx71YgYVBYskUWIzoSnxDh+TjlWbT"
+    "EZmO0F+G8Z79fIbwif3u9nvDdyzBo2vdxJ3WY3CiAJpUGWWrdWmwXEZOXcsqXMvKNu8Sd6"
+    "yKGZQTbMiyrWLmFCGoYubUwBYyvclrN55W/TyjKKj43hQmW6B8mx9Fl1OROvG91wSjP/BY"
+    "l9C8SdZR6QffaSHzXzxWpG7zSd2q56nUMar8kdNUyyrY1YKYsqqlVrVPsOdXom0XEk3R0X"
+    "1jqjZy1EZOYzdy1vxqz0G+19OYU5QkQmu4t4KoCkmrQ9xw2eZDWQT4ZrsOh96V28HHbNTp"
+    "YsWUKkJNMaU/3sBW/XbNLrmqFiSONdElVFWcc1TGVIFFmdrQVMX3/Kx4WicesEPGvG7njE"
+    "QxK8W+WiSlU4qtmJSI8k4XYdj01agAYly8mQDu5IPGhXGDxd8zLI4b3Nu3DHdmD27tW4YH"
+    "XV6+/x9WSMI2"
+)

migrations/models/13_20260917193429_add_beli_profiles_restaurants.py

@@ -0,0 +1,131 @@
+from tortoise import BaseDBAsyncClient
+
+RUN_IN_TRANSACTION = True
+
+
+async def upgrade(db: BaseDBAsyncClient) -> str:
+    return """
+        CREATE TABLE IF NOT EXISTS "cron_jobs" (
+    "id" UUID NOT NULL PRIMARY KEY,
+    "name" VARCHAR(255) NOT NULL,
+    "description" TEXT,
+    "prompt" TEXT NOT NULL,
+    "recipient" VARCHAR(255) NOT NULL,
+    "channel" VARCHAR(20) NOT NULL,
+    "enabled" BOOL NOT NULL DEFAULT True,
+    "last_run_at" TIMESTAMPTZ,
+    "last_status" VARCHAR(50),
+    "last_error" TEXT,
+    "created_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updated_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+COMMENT ON COLUMN "cron_jobs"."channel" IS 'IMESSAGE: imessage\nEMAIL: email';
+        CREATE TABLE IF NOT EXISTS "beli_profiles" (
+    "id" UUID NOT NULL PRIMARY KEY,
+    "beli_email" VARCHAR(255) NOT NULL,
+    "beli_password" TEXT NOT NULL,
+    "beli_user_id" VARCHAR(255),
+    "beli_refresh_token" TEXT,
+    "last_refreshed_at" TIMESTAMPTZ,
+    "last_error" TEXT,
+    "enabled" BOOL NOT NULL DEFAULT True,
+    "created_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updated_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "user_id" UUID NOT NULL REFERENCES "users" ("id") ON DELETE CASCADE,
+    CONSTRAINT "uid_beli_profil_user_id_f4bba0" UNIQUE ("user_id", "beli_email")
+);
+        CREATE TABLE IF NOT EXISTS "beli_restaurants" (
+    "id" UUID NOT NULL PRIMARY KEY,
+    "beli_business_id" VARCHAR(255) NOT NULL,
+    "name" VARCHAR(500) NOT NULL,
+    "city" VARCHAR(255),
+    "neighborhood" VARCHAR(255),
+    "cuisine" JSONB NOT NULL,
+    "price_range" VARCHAR(10),
+    "status" VARCHAR(20) NOT NULL,
+    "rank_index" INT,
+    "score" DECIMAL(5,2),
+    "notes" TEXT,
+    "tags" JSONB NOT NULL,
+    "fetched_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "created_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "profile_id" UUID NOT NULL REFERENCES "beli_profiles" ("id") ON DELETE CASCADE,
+    CONSTRAINT "uid_beli_restau_profile_921fe6" UNIQUE ("profile_id", "beli_business_id", "status")
+);
+COMMENT ON COLUMN "beli_restaurants"."status" IS 'BEEN: been\nWANT_TO_TRY: want_to_try';"""
+
+
+async def downgrade(db: BaseDBAsyncClient) -> str:
+    return """
+        DROP TABLE IF EXISTS "beli_profiles";
+        DROP TABLE IF EXISTS "cron_jobs";
+        DROP TABLE IF EXISTS "beli_restaurants";"""
+
+
+MODELS_STATE = (
+    "eJztXe9zmzgT/lcYf+rN5DJx0vTuMu/cjJ3Q1m1sd2Lner26w8ggx1xAcPxo6un0f38lAT"
+    "YIgY1tbJTqw11qoRXwaCXtPrsS31u2Y0DLP7120Ffo+SAwHdS6Ur63ELAh/gf3+onSAq67"
+    "ukoKAjC1qICeqkmvgKkfeEAP8MUZsHyIiwzo657pxjdDoWWRQkfHFU30sCoKkflfCLXAeY"
+    "DBHHr4wucvuNhEBvwG/eSn+6jNTGgZmec2DXJvWq4FC5eW3d/3bl7TmuR2U013rNBGq9ru"
+    "Ipg7aFk9DE3jlMiQaw8QQQ8E0Ei9BnnK+LWTouiJcUHghXD5qMaqwIAzEFoEjNb/ZiHSCQ"
+    "YKvRP538s/WxXgwVATaE0UECy+/4jeavXOtLRFbnX9tnP34uLVL/QtHT948OhFikjrBxUE"
+    "AYhEKa4rIOnfHJTXc+DxoUzqM2DiB90GxqRgheNKhxIgE4C2Q61lg2+aBdFDMMc/zy8vS2"
+    "D8q3NHkcS1KJQO1utI6wfxpfPoGoF0BaHuQfLKGgjyQN7gK4FpQz6YWUkGUiMWPU3+0VCA"
+    "8TsYQ2Qt4kFQgu+411dH407/A3kT2/f/syhEnbFKrpzT0gVT+uIV0xXLRpSPvfFbhfxU/h"
+    "kOVFb3l/XG/7TIM4EwcDTkPGnASI3XpDQBJtOxoWts2bFZSdmxR+3Y+OFTA3YOEIJWlWkv"
+    "JbLVzBc/VaXuaz3B6Q7rBTPznW0y8Z0Vz3tn7LQX+tDTqq3DKZEdFuNtsNxtKGy/9BIDZv"
+    "bIXXkJGHnwXjseNB/Qe7igEPbwcwCk85bc2HS7j5tpKmir0tVTeOBpadOllQK/HX4nGESD"
+    "sTO67tyoLYrhFOiPT8AztAIwbej74AH6eUC7seTr93fQWhq4fCzTZnA/arGZc3MRthQr59"
+    "xJYZRBL3/JPrfZEoDwexvxvcmdStBZ40ukQNzMpdDSHSldC6FdiwB+4xhOY1zKhzKpL4pr"
+    "UWYQqX+PM7ZQspC+6Hf+/iVjD90OB2+S6qmF9/p22JV+xnM0R/N+hu9C8MgzBog9qqLQzp"
+    "kCmd5NiR956LTuR+rdlUJW9Aka9frdzpXim/YUTNB4OLzFaDgOtaGrWq/tTazXdrH12s5Z"
+    "r6aNVxntES6qeAEZoYP5AfskQC7b5xsgiWsVQkmvZbGM12wN+0iGVYlNyksKiWo9tFLaLK"
+    "pmlHBE92mhCOpw6QzJvaPjxXLmjUVxrQfGUZeqnlidfgfxcPvQdrxFi+NupK6WehnUzbRJ"
+    "RVN6F8/Au8A3CSCq5GCkRKSPIX2Mn8rHkLGMZ9GxuVjG0Vj4Zk2GotDwDbYCd+Dh67b+iu"
+    "y+9RaftPTEt/RIP1ZNU0nLiGLrHYBTcIHvPzkeRzO7JgLegg9mWooBc7oIIm+qcQxNCXhd"
+    "YvemV3cMWbc36Nx94pvOSf203nY/jdUOgy60gVkpq2ApsB8VXT/i96mg7bPNSNkyVjZHyz"
+    "7NQeDjKV1DoT0tosT5YHJE90ImHhbVi01AvSjG9ILDdMcka3VIOaIS0mTcahCR1+PNpY5j"
+    "QYBKxnxalp1R4xBJHetTVeunwpQ6HN5mptRuj2Ub7vtdFc8HFGZcyYysy95gzAV3bgMdm1"
+    "2PkEPVrplQGVkBNbb9apOZlXU6UxPrK1ZjHdPQNT+c/gt1DgFQDCgrJyCYtRhS2CWfa67n"
+    "fDWNalNqTvBw1mnLcnRg7WDeM6HETabVy+Jp9TI3rVoGcLUHzwldTh7Xu9FwwIeUEWMAvU"
+    "f4LT8bph6cKJbpB19qg/fzl3pcJ/Lm5UQvy+kytippQBK9z5IPlETvM+3YZTZnjmhblwqb"
+    "26G1n3zYJjv5+TzYbHrKKui7PRbZOHPzNH8jKCL7GOi6ExKxnQBRSVudqCmBIfH1OTRC7I"
+    "tp+8khHyXtiZlAnoZmCi2T2Koz09oVlS5u6kPUkmCA1BnboEOoxQluqAkxWBzdoEN5s/BG"
+    "izanxPqN/waAkO0KQIYS50ScthiUNpGZoAkaBQ5e4hQXeD40FPpQCq34ZAZz5eLsVwMsFA"
+    "+S+vhhTpVbEz36SuAo13PPscFNd4Jmjqd8xf4l/uND4OlzRQcumJqWGeBZmz6ZDMNsoct7"
+    "DsMkXCAP0PUZnlxgheC5a+EPtqBidmVhjh7RutwoYHBZEjC4zAcMZngWIUY01mPOElWMJi"
+    "snJqS16Cb2QQoBLdm944gI56Hz64j3XNUvT2TE9MgF8cCT1y7lVqaOsdCqbmvLCAmyveHQ"
+    "w4JiNA9sTvR8DbCJkASWn89LrWx8R/xoFS03jqggIMszXyTLui19Dr+5Jm5ti47NSorZsY"
+    "J05EaLdcxyVkyOzkrJ/OgEkDyGlVOkxSWMT5hU6aySNClbOoNxEbGY6oM1/GImVLApzxjL"
+    "EJpwZj6EHqWGFcLy9fqdD6QYQUp3+UWsY4UWUhzkUsqDBqEb8TNGLCREurdwsaIpSSbrSa"
+    "oNxYdBgG/vn0wQYTeTZrA6B6F/qoxC13W8wFdsrGKma8Gkhq+40KOb3SVH2RCOMlba6nRQ"
+    "TlAylQljYfquBRZa1Qx8Vk76DqmjJ1xtjhW7CpwZIVHYtUOBSebnPJg9VEAcZGQYLM36bJ"
+    "M//rjYAUp62Nav5+2Xv738/eLVy99xFfooy5LfSsDNZ9lSDLbZWZMTlMrIKGPhHhs1sUOK"
+    "Oa1cC6Kge2hiy8SONrZHvnJ0tzT9PiN3wNT75cLf4Mx7C/iBBj3P4WQ0F6tsVkqQZV6eqy"
+    "D5QpluKztWnqsgz1Vo7rkKDLGxj4RdsbDM5kwtkK5FrNwegBjh1kbLxgSCpHb6OIVMEYOc"
+    "BW8dicx03HoeeYwvP/oKkVNczyGTjk8zT0kjkNKuME0U53nkyi0QHrlPYMb/+QqxaSNhsv"
+    "ZiGds9iQpxYzpuCxrL1Fg8E8bM8Qw3GHpQIW/3iB+EpLn6EXusmNj3gzbE7VtRw0SAWs0K"
+    "PfwSV5csckNYZOrRUKXdJl0rL70H265ZHlKDTLmNQsG0S5JM5JA3Tgq5Qp7o4SjDsx1M8z"
+    "0ThgkGBdHgQgBzcj8leqR5qIeEeNLiZYJjxBSCWCT+U2JJh2SMwvYzNNuAnKSPPElLJuV5"
+    "MikyD+uZ5WHVndQgfhZWbg8wx4/m7RMudqT5u5TlwZZC+5ke1E3X5B5iXhx/zwiJEh0+RE"
+    "Z+2Ucd139EZ9cPPO4P1BZZcUedN+qVkpy/N0Fqv9O7vVKWB0Ye//OP8vz9OuLER9gX/Cxz"
+    "7FbrZXV3gpUV06EQxIHYyDUsCvps+H20pfQBz9qLrbooSHL6QR3c9AZv8hZIK75ypbgQGb"
+    "g/J2ikDsZXio8nygl6jWd99eaKRhagMUHXncG1ekuLdPKuVnyG6PGXA2yWhJ4HY/y36aZs"
+    "CwfsqrvljU8HsQIznUSKMbAOwivxDe6ST1eKgXtkMUEfVfU9+fkE4SP53R8Oxm9JgY3Xur"
+    "m1aEbnhD7UsDLyVuvSZLmMnDynljmnlgTvEnesihmUExRk2ZY5c5IQlDlzsmMLmd5k2E0X"
+    "Vb9XyQpKvjeFyR4oX/Gz6HIq0iS+99pz0Dtn2uLQvMmlkzJ2V8eVtH+dqSR1xSd1q+6nkt"
+    "uo8ltOU09Wwa5mxKRVzbWqXc+x3Uq07UpCFB09NKYykCMDOcIGcrb8jNFRPmAkzC5KL0Rb"
+    "uLeMqExJa0LecFnwoSwDfLeow7GjcjV83UfuLpZMqSTUJFP683Vs1Y/51MlVpb/EweGrmA"
+    "91FHNWuY+D7Je3+rzcQExvFNnLXySbtc2aswOblQK/ggGUlZK+LfNFnV0OCMq1IAq6h7aJ"
+    "KFCFpzis0d3ioxyaanIeSnk9OMOzx7zo07ybanCuGUFgPrQWR4xABNZWdiC3AUksNIFYkL"
+    "5wHSNG0pn7pTMlt/AsXFDJLTzTjpUnl8mTy5p8cplHjk8KPbDz94YJM3W3bEwsZGs9tIsB"
+    "poDUy0K3htdjOm3f1J67Ihjp7aahbyLo+7GyxaEzyfYdhe1jOqMSb8LIisJNHYA8+bnSAm"
+    "v51KtuBpxc6JIP2MX1BXGUD6GE2IKZTx1v7jiVxjYrJxFd6mRokhkvD+a70XBQoJYrEQbH"
+    "e4Rf8LNh6sGJYpl+8KWuoZ5ajqahaQUm8k/JDWtakQgW5TwPS+kwSxdpIJ/HaupQwyYKb/"
+    "NdsSozYkJqcnuTybVdPLe2c1OreBus+RmXXVUdXClTCNEEfewMxtp4qI3vyP5cbMpiI1QL"
+    "vIZsycUa+KjRF8yjXngMYFZoq8P/jqC7ez7+z9exJ86hrKBu2sDiA7eUYcmqSOg0Fm4kfi"
+    "Xg3KjXvX7n9sXlyTnD4yY6+TKneMgJeEdPFkcdlgKCzJWHDjgE4IGDZ/H6n9SXi//2i/8M"
+    "Bvp28dCspOSum8Vdy2DTs+jXfLApZvwqxiWyUjI0kSZOd4xOMHmejcVvbZAiqyNN2hXegd"
+    "jXnLc4fHx85aSMhwerOo3ZEF78Rc0Nz8WPO3A3Mr0J7kgxh/4Vej5343IxJ5ESEZPurYVa"
+    "I0OjAohxdTEBbG/El7dL+PI2hy8vOqGzhJssPKHzYO5Jbavu3hyRo27k+PF/70kbqw=="
+)

migrations/models/14_20260917205441_beli-update.py

@@ -0,0 +1,89 @@
+from tortoise import BaseDBAsyncClient
+
+RUN_IN_TRANSACTION = True
+
+
+async def upgrade(db: BaseDBAsyncClient) -> str:
+    return """
+        ALTER TABLE "beli_restaurants" DROP CONSTRAINT IF EXISTS "uid_beli_restau_profile_921fe6";
+        ALTER TABLE "beli_restaurants" ADD "friend_name" VARCHAR(255);
+        ALTER TABLE "beli_restaurants" ADD "friend_beli_user_id" VARCHAR(255) NOT NULL DEFAULT '';
+        CREATE UNIQUE INDEX IF NOT EXISTS "uid_beli_restau_profile_1e4068" ON "beli_restaurants" ("profile_id", "beli_business_id", "status", "friend_beli_user_id");"""
+
+
+async def downgrade(db: BaseDBAsyncClient) -> str:
+    return """
+        DROP INDEX IF EXISTS "uid_beli_restau_profile_1e4068";
+        ALTER TABLE "beli_restaurants" DROP COLUMN "friend_name";
+        ALTER TABLE "beli_restaurants" DROP COLUMN "friend_beli_user_id";
+        CREATE UNIQUE INDEX IF NOT EXISTS "uid_beli_restau_profile_921fe6" ON "beli_restaurants" ("profile_id", "beli_business_id", "status");"""
+
+
+MODELS_STATE = (
+    "eJztXe9zmzgT/lcYf+rN5DJx0vTuMu/cjJ3Q1m1sd2Lner26w8ggx1xAcPxo6un0f38lAT"
+    "YIgY1tbJTqw11qoRXwaJF2n11J31u2Y0DLP7120Ffo+SAwHdS6Ur63ELAh/gf3+onSAq67"
+    "ukoKAjC1qICeqkmvgKkfeEAP8MUZsHyIiwzo657pxjdDoWWRQkfHFU30sCoKkflfCLXAeY"
+    "DBHHr4wucvuNhEBvwG/eSn+6jNTGgZmec2DXJvWq4FC5eW3d/3bl7TmuR2U013rNBGq9ru"
+    "Ipg7aFk9DE3jlMiQaw8QQQ8E0Ei9BnnK+LWTouiJcUHghXD5qMaqwIAzEFoEjNb/ZiHSCQ"
+    "YKvRP538s/WxXgwVATaE0UECy+/4jeavXOtLRFbnX9tnP34uLVL/QtHT948OhFikjrBxUE"
+    "AYhEKa4rIOnfHJTXc+DxoUzqM2DiB90GxqRgheNKhxIgE4C2Q61lg2+aBdFDMMc/zy8vS2"
+    "D8q3NHkcS1KJQO1utI6wfxpfPoGoF0BaHuQfLKGgjyQN7gK4FpQz6YWUkGUiMWPU3+0VCA"
+    "8TsYQ2Qt4o+gBN9xr6+Oxp3+B/Imtu//Z1GIOmOVXDmnpQum9MUrpiuWjSgfe+O3Cvmp/D"
+    "McqKzuL+uN/2mRZwJh4GjIedKAkfpek9IEmEzHhq6xZcdmJWXHHrVj44dPfbBzgBC0qgx7"
+    "KZGtRr74qSp1X+sJTneYL5iR72yTge+seNw7Y4e90IeeVm0eTonsMBlvg+Vun8L2Uy8xYG"
+    "aP3JmXgJEH77XjQfMBvYcLCmEPPwdAOm/KjU23+7iZpoK2Kl09hQeeljZdWinw2+F3gkH0"
+    "MXZG150btUUxnAL98Ql4hlYApg19HzxAPw9oN5Z8/f4OWksDl49l2gzuRy02c2wuwpZi5Z"
+    "w7KYwy6OUv2ec2WwIQfm8jvje5Uwk6a3yJFIibuRRauiOlayG0axHAbxzDaYxL+VAm9UVx"
+    "LcoMIvXvccYWSibSF/3O379k7KHb4eBNUj018V7fDrvSz3iO5mjez/BdCB55xgCxR1UU2j"
+    "lTINO7KfEjfzqt+5F6d6WQGX2CRr1+t3Ol+KY9BRM0Hg5vMRqOQ23oqtZrexPrtV1svbZz"
+    "1qtp41lGe4SLKl5ARuhgfsA+CZDL9vkGSOJahVDSa1ks4zlbwz6SYVVik/KSQqJaD62UNo"
+    "uqGSUc0X1aKII6XDpDcu/oeLGceWNRXOuBcdSlqidWp99BPNw+tB1v0eK4G6mrpV4GdTNt"
+    "UtGU3sUz8C7wTQKIKjkYKRHpY0gf46fyMWQs41l0bC6WcTQWvlmDoSg0fIOtwB14+Lqtvy"
+    "K7b73FJy098S090o9V01TSMqLYegfgFFzg+0+Ox9HMromAt+CDmZZiwJwugsibahxDUwJe"
+    "l9i96dkdQ9btDTp3n/imc1I/rbfdT2O1w6ALbWBWyipYCuxHRdd/8ftU0PbZZqRsGSubo2"
+    "Wf5iDw8ZCuodCeFlHifDA5onshEw+L6sUmoF4UY3rBYbpjkrU6pBxRCWny3WoQkdfjjaWO"
+    "Y0GASr75tCw7osYhkjrmp6rWT4UhdTi8zQyp3R7LNtz3uyoeDyjMuJIZWZe9wZgL7twGOj"
+    "a7HiGHql0zoDKyAmps+9UmIyvrdKYG1lesxjqmoWt+OP0X6hwCoBhQVk5AMGsxpLBLPtdc"
+    "z/lqGtWG1Jzg4azTluXowNrBvGdCiZsMq5fFw+plbli1DOBqD54Tupw8rnej4YAPKSPGAH"
+    "qP8Ft+Nkw9OFEs0w++1Abv5y/1uE7kzcuJXpbTZWxV0oAkep8lHyiJ3mfasctszhzRti4V"
+    "NrdCaz/5sE128vN5sNn0lFXQd3sssnHm5mn+RlBE9jHQdSckYjsBopK2OlFTAkPi63NohN"
+    "gX0/aTQz5K2hMzgTwNzRRaJrFVZ6a1Kypd3NSHqCXBAKkztkE/oRYnuKEmxGBxdIN+ypuF"
+    "N1q0OSXWb/w3AIRsVwAylDgn4rTFoLSJzARN0Chw8BSnuMDzoaHQh1JoxSczmCsXZ78aYK"
+    "F4kNTHD3Oq3Jro0VcCR7mee44NbroTNHM85Sv2L/EfHwJPnys6cMHUtMwAj9r0yWQYZgtd"
+    "3nMYJuECeYCuz/DkAisEz10Lf7AFFbMrC3P0iNblRgGDy5KAwWU+YDDDowgxorEec6aoYj"
+    "RZOTEhrUU3sQ9SCGjJ6h1HRDgPnV9HvOeqfnkiI6ZHLogHnrx2KbcydYyFVnVZW0ZIkOUN"
+    "h/4sKEbzwOZEz9cAmwhJYPn5vNTKxnfEj1bRcuOICgKy3PNFsqzb0ufwm2vi1rbo2KykmB"
+    "0rSEduNFnHLGfF5OislMyPTgDJY1g5RVpcwviESZXOKkmTsqUzGBcRi6k+WMMvZkIFm/KM"
+    "sQyhCWfmQ+hRalghLF+v3/lAihGkdJdfxDpWaCHFQS6lPGgQuhE/Y8RCQqR7CxcrmpJksp"
+    "6k2lB8GAT49v7JBBF2M2kGq3MQ+qfKKHRdxwt8xcYqZroWTGr4igs9uthdcpQN4Shjpa1O"
+    "B+UEJVOZMBam71pgoVXNwGflpO+Q2nrC1eZYsavAmREShV07FJhkfM6D2UMFxEFGhsHSrM"
+    "82+eOPix2gpJtt/Xrefvnby98vXr38HVehj7Is+a0E3HyWLcVgm5U1OUGpjIwyFq6xURM7"
+    "pJjTyrUgCrqHJrZM7Ghje+QrR3dL0+8zcgdMvV9O/A3OvLeAH2jQ8xxORnOxymalBJnm5b"
+    "4Kki+U6bayY+W+CnJfhebuq8AQG/tI2BULy2zO1ALpWsTK7QGIEW5ttGxMIEhqp49TyBQx"
+    "yFnw1pHITMet55HH+PKjrxA5xfUcMuj4NPOUNAIp7QrTRHGeR67cAuGR+wRm/J+vEJs2Ei"
+    "ZzL5ax3ZOoEDem47agsUyNxSNhzBzPcIOhBxXydo/4QUiaqx+xx4qJfT9oQ9y+FTVMBKjV"
+    "rNDNL3F1ySI3hEWmHg1V2m3StfLSe7DtmuUhNciU2ygUTLskyUQOed9JIVfIEz0cZXi2g2"
+    "m+Z8IwwaAgGlwIYE7up0SPNA/1kBBPWjxNcIyYQhCLxH9KLOknGaOw/QjNNiAH6SMP0pJJ"
+    "eZ5MiszDemZ5WHUnNYifhZVbA8zxo3nrhIsdaf4qZbmxpdB+pgd10zW5m5gXx98zQqJEhw"
+    "+RkV92qOP6Q3R2PeBxf6C2yIw76rxRr5Rk/70JUvud3u2Vstww8vjHP8r99+uIEx9hXfCz"
+    "zLFbzZfV3QlWVkyHQhAHYiPXsCjos+H5aEvpA+61F1t1UZDk9IM6uOkN3uQtkFZ85UpxIT"
+    "Jwf07QSB2MrxQfD5QT9BqP+urNFY0sQGOCrjuDa/WWFunkXa14D9HjTwfYLAk9D8b4b9NN"
+    "2RYO2FV3yxufDmIFZjqJFGNgHYRn4hvcJZ+uFAP3yGKCPqrqe/LzCcJH8rs/HIzfkgIbz3"
+    "Vza9GMzgl9qGFl5M3WpclyGTm5Ty2zTy0J3iXuWBUzKCcoyLQtc+YkIShz5mTHFjK9yWc3"
+    "XVQ9r5IVlHxvCpM9UL7iZ9HlVKRJfO+156B3zrTFoXmTSydl7K6OK2n/OlNJ6opP6lZdTy"
+    "WXUeWXnKaerIJdzYhJq5prVbueY7uVaNuVhCg6emhMZSBHBnKEDeRseYzRUQ4wEmYVpRei"
+    "LdxbRlSmpDUhb7gs+FCWAb5b1OHYUbkaTveRq4slUyoJNcmU/nwdW/Uwnzq5qvRJHBy+ij"
+    "moo5izyh0Osl/e6vNyATG9UWQvf5Fs1jZzzg5sVgr8CgZQVkr6tsyJOrtsEJRrQRR0D20T"
+    "UaAKd3FYo7vFWzk01eQ8lPJ6cIZHj3nR0bybanCuGUFgPrQWR4xABNZWdiC3AUksNIFYkL"
+    "5wHV+MpDP3S2dKbuFZuKCSW3imHSt3LpM7lzV55zKPbJ8UemDn84YJM3W3bEwsZGvdtIsB"
+    "poDUy0K3htdjOm3f1J67Ihjp7aahbyLo+7GypSJungmRoWV8ckkCHoUEZPqoEp3CyIpCWR"
+    "2AU+EpeAV0C8QPuHhqB2U9HLxVczIZMUE87QPg+XMlt9ZyYLFuBpyM/pJjGOP6UgmXSojt"
+    "8PnU8eaOU2mwZOUkokudDE0yQefBfDcaDgrUciXC4HiP8At+Nkw9OFEs0w++1Db3rKynaW"
+    "hagYn8U3LDmgwogkU5W8kSk4ylRRrIZ2ObOtSwoc1bQlqsyoyYkJrc3mRwbRePre3c0Cre"
+    "NgH8vOGuqg6ulCmEaII+dgZjbTzUxndklTl2yLArpQVeQxaWYw181OgL5lEv3MwyK7TVFp"
+    "ZH0N09b2Lp647H27gS6qYNLD5wSxmWco2ETmPhRuJXAs6Net3rd25fXJ6cM9GIRCdf5hQP"
+    "OQFvA9Xi2NlSQJCx8tBhswA8cPAsnv+T+nLy337yn8FA3y6qn5WUEZhmRWBkyPRZ9Gs+ZB"
+    "rz1hWja1kpGWBL0/87xtiYbOXG4rc21JbVkSbtbdCB2NectzhRpfjKSVk0CazqNGZbg+Jz"
+    "YTc83SHuwN1iP01wR4pDPl+h53OX3xdzEikRMeneWqg18mlUADGuLiaA7Y348nYJX97m8O"
+    "VF+8yWcJOF+8wezD2pbdbdmyNy1OVIP/4PbwAAKw=="
+)

migrations/models/15_20260917220930_beli_add_beli_category_09172026.py

@@ -0,0 +1,87 @@
+from tortoise import BaseDBAsyncClient
+
+RUN_IN_TRANSACTION = True
+
+
+async def upgrade(db: BaseDBAsyncClient) -> str:
+    return """
+        DROP INDEX IF EXISTS "uid_beli_restau_profile_1e4068";
+        ALTER TABLE "beli_restaurants" ADD "beli_category" VARCHAR(10) NOT NULL DEFAULT '';
+        CREATE UNIQUE INDEX IF NOT EXISTS "uid_beli_restau_profile_93dc73" ON "beli_restaurants" ("profile_id", "beli_business_id", "status", "friend_beli_user_id", "beli_category");"""
+
+
+async def downgrade(db: BaseDBAsyncClient) -> str:
+    return """
+        DROP INDEX IF EXISTS "uid_beli_restau_profile_93dc73";
+        ALTER TABLE "beli_restaurants" DROP COLUMN "beli_category";
+        CREATE UNIQUE INDEX IF NOT EXISTS "uid_beli_restau_profile_1e4068" ON "beli_restaurants" ("profile_id", "beli_business_id", "status", "friend_beli_user_id");"""
+
+
+MODELS_STATE = (
+    "eJztXW1v27YW/iuCP3VAFsRJ023BxQA7cVu3sV3EzrquLgRaomMtEqXppalR9L9fkpJsiS"
+    "Jly69iyg9baoqHkh4ekec855D83nBcE9rB6bWLvkI/AKHlosaV9r2BgAPxP7jXT7QG8Lzl"
+    "VVIQgolNBYxMTXoFTILQB0aIL06BHUBcZMLA8C0vuRmKbJsUugauaKGHZVGErP8iqIfuAw"
+    "xn0McXPn/BxRYy4TcYpD+9R31qQdvMPbdlknvTcj2ce7Ts/r5785rWJLeb6IZrRw5a1vbm"
+    "4cxFi+pRZJmnRIZce4AI+iCEZuY1yFMmr50WxU+MC0I/gotHNZcFJpyCyCZgNP43jZBBMN"
+    "Doncj/Xv7ZqAAPhppAa6GQYPH9R/xWy3empQ1yq+u3rbsXF69+oW/pBuGDTy9SRBo/qCAI"
+    "QSxKcV0CSf8WoLyeAZ8PZVqfARM/6CYwpgVLHJc6lAKZArQZag0HfNNtiB7CGf55fnlZAu"
+    "NfrTuKJK5FoXSxXsda308uncfXCKRLCA0fklfWQVgE8gZfCS0H8sHMSzKQmonoafqPmgKM"
+    "38EcIHuefAQl+I66vc5w1Op9IG/iBMF/NoWoNeqQK+e0dM6UvnjFdMWiEe1jd/RWIz+1fw"
+    "b9Dqv7i3qjfxrkmUAUujpyn3RgZr7XtDQFJtexkWdu2LF5SdWxR+3Y5OEzH+wMIATtKsNe"
+    "RmSjkS95qkrd13iCky3mC2bkO1tn4DsTj3tn7LAXBdDXq83DGZEtJuNNsNzuU9h86iUGzP"
+    "SRO/MSMIrgvXZ9aD2g93BOIezi5wDI4E25iel2nzRTV9CWpcun8MHTwqbLKgV+O/xOMIw/"
+    "xtbwunXTaVAMJ8B4fAK+qQvAdGAQgAcYFAFtJ5Kv399Be2Hg8rHMmsG9uMV6js0ibClW7r"
+    "mbwSiHXvGSc+6wJQDh9zaTe5M7laCzwpfIgLieS6FnO1K5FlK7FiH8xjGcRriUD2VaXxbX"
+    "oswg6vw9ytlC6UT6otf6+5ecPXQ76L9Jq2cm3uvbQVv5Gc/RHC36GYEHwSPPGCD2aAdFTs"
+    "EUyPVuRvzIn07jfti5u9LIjD5Gw26v3brSAsuZgDEaDQa3GA3XpTZ0Veu1uY712hRbr82C"
+    "9Wo5eJbRH+G8iheQEzqYH7BLAuSyeb4GkriWEEp6LY9lMmfr2Ecy7UpsUlFSSlT3Qytlza"
+    "JqRglHdJcWiqQOl8GQ3Fs6XixnXlsUV3pgHHWp6ont0+8gHm4POq4/b3DcjczVUi+DupkO"
+    "qWgp7+IZeBf4JiFElRyMjIjyMZSP8VP5GCqW8Sw6thDLOBoLX6/BUBYavsZW4BY8/L6tP5"
+    "Hdt9riU5ae/JYe6ceqaSpZGVlsvQNwCh4IgifX52hm20LAn/PBzEoxYE7mYexN1Y6hKQGv"
+    "Teze7OyOIWt3+627T3zTOa2f1dv2p1GnxaALHWBVyipYCOxGRVd/8btU0ObZeqRsGStboG"
+    "WfZiAM8JCuo8iZiChxPpgc0Z2QiYdF9WIdUC/EmF5wmO6EZK0OKUdUQZp+tzpE5PV4Y6nr"
+    "2hCgkm8+K8uOqEmIZB/zU1Xrp8KQOhjc5obUdpdlG+577Q4eDyjMuJIVW5fd/ogL7swBBj"
+    "a7HiGHql0xoDKyEmps89U6IyvrdGYG1lesxrqWaehBNPkXGhwCQAwoKychmHsxpLBLPtM9"
+    "3/1qmdWG1ILg4azThu0awN7CvGdCiesMq5fiYfWyMKzaJvD0B9+NPE4e17vhoM+HlBFjAL"
+    "1H+C0/m5YRnmi2FYRf9gbv5y/7cZ3Im5cTvSyny9iqpAFF9D5LPlARvc+0YxfZnAWibVUq"
+    "bGGF1m7yYevs5BfzYPPpKcug7+ZY5OPM9dP8taCI7WNgGG5ExLYCpEPaasVNSQxJYMygGW"
+    "FfTN9NDvkwbU/OBPIsNBNoW8RWnVr2tqi0cVMf4pYkA2SfsQ36CTU4wY1OSgyKoxv0U14v"
+    "vNGgzWmJfuO/ISBkuwaQqSU5EacNBqV1ZMZojIahi6c4zQN+AE2NPpRGKz5Z4Uy7OPvVBH"
+    "PNh6Q+fphT7dZCj4EWutr1zHcdcNMeo6nra1+xf4n/BBD4xkwzgAcmlm2FeNSmT6bCMBvo"
+    "8o7DMCkXyAN0dYYnF1gpeO698AcbUDHbsjBHj2hdrhUwuCwJGFwWAwZTPIoQIxrrMWeKEq"
+    "PJyskJ6V50E/sgQkBLVu+4MsJ56Pw64j1X9ctTGTk9ckk88PS1S7mViWvO9arL2nJCkixv"
+    "OPRnQTGahQ4ner4C2FRIAcvP56VWNr4jfrSKlhtHVBKQ1Z4vimXdlD6H3zwLt7ZBx+Yl5e"
+    "xYSTpyrck6YTkrJkfnpVR+dApIEcPKKdLyEsYnTKp0XknqlC2dw1hELGb6YAW/mAsVrMsz"
+    "JjKEJpxaD5FPqWGNsHzdXusDKUaQ0l2BiHWs0EKGg1xI+dAkdCN+xpiFhMjw5x5WNC3NZD"
+    "3JtKEFMAzx7YOTMSLsZtoMVucwCk61YeR5rh8GmoNVzPJsmNYINA/6dLG74ihrwlEmSlud"
+    "DioIKqYyZSyswLPBXK+agc/KKd8hs/WEp8+wYleBMyckC7t2KDDJ+FwEs4sExEFOhsHS2p"
+    "9t8scfF1tASTfb+vW8+fK3l79fvHr5O65CH2VR8lsJuMUsW4rBJitrCoJKGRllFK6x6aR2"
+    "iJjTKrQgC7qHJrYs7Ghje+QrR3dL0+9zcgdMvV9M/DXOvLdBEOrQ911ORrNYZfNSkkzzal"
+    "8FxReqdFvVsWpfBbWvQn33VWCIjV0k7MqFZT5nao4MPWbldgDEELc2XDQmESR7p48zyIgY"
+    "5Dx4q0hkpuNW88gjfPkx0Iic5vkuGXQCmnlKGoGUdoVZorjII1dugfDIPQIz/i/QiE0bC5"
+    "O5F8s43klciBszcFvQXKTG4pEwYY6nuMHIhxp5u0f8ICTNNYjZY83Cvh90IG7fjhsmAtRq"
+    "1ujml7i6YpFrwiJTj4Yq7SbpWkXpHdh29fKQamTKrRUKpl2SZiJHvO9EyBXyRA9HGZ5tYZ"
+    "rvmDBMMRBEg4UAFuR+SvRI89CICPGkJ9MEx4gRgigS/ymxpJ9kgsLmIzTbgBqkjzxIKybl"
+    "eTIpKg/rmeVh7TupQf4srMIaYI4fzVsnLHak+auU1caWUvuZPjQsz+JuYi6Ov+eEZIkOHy"
+    "Ijv+xQx9WH6Gx7wOPuQG2QGXfYetO50tL998ao02t1b6+0xYaRxz/+Ue2/v4848RHWBT/L"
+    "HLvlfFndnWBl5XQoJHEg1nINRUGfNc9HW0gfcK+9xKqLgySnHzr9m27/TdECaSRXrjQPIh"
+    "P35xgNO/3RlRbggXKMXuNRv3NzRSML0Byj61b/unNLiwzyrnayh+jxpwNslkS+DxP8N+mm"
+    "fAsH7Kq7xY1P+4kCM51EijGwLsIz8Q3ukk9Xmol7ZD5GHzud9+TnE4SP5Hdv0B+9JQUOnu"
+    "tm9rwenRMFUMfKyJutS5PlcnJqn1pmn1oSvEvdsSpmUEFQkmlb5cwpQlDlzKmOFTK96Wc3"
+    "mVc9r5IVVHxvBpMdUL7yZ9EVVKROfO+176J37qTBoXnTSydl7K6BK+n/uhNF6spP6lZdT6"
+    "WWURWXnGaerIJdzYgpq5prVXu+63iVaNulhCw6emhMVSBHBXKkDeRseIzRUQ4wkmYVpR+h"
+    "DdxbRlSlpNUhb7gs+FCWAb5d1OHYUbk9nO6jVhcrplQRaoop/fk6tuphPvvkqrIncXD4Ku"
+    "agDjFnVTgcZLe81efFAmJ6o9he/qLYrE3mnC3YrAz4FQygvJTybZkTdbbZIKjQgizoHtom"
+    "okAJd3FYobvirRzqanIeSnl9OMWjx0x0NO+6GlxoRhKYD63FMSMQg7WRHchtQBELdSAWlC"
+    "+8jy9G0Zm7pTMVt/AsXFDFLTzTjlU7l6mdy+q8c5lPtk+KfLD1ecOEmbpbNCYXsnvdtIsB"
+    "RkDq5aFbwesxnbZras9bEoz0dpMosBAMgkTZMhE334LI1FmfnP42cM8/kOO4FSt4FFaQ6b"
+    "RK/AojKwuHdQCSRaDx66IrED/gaqotlPVw8FZN0mTEJHG9D8UJLkbiqsNAVlBGFW2ukyTR"
+    "FCdJNAtJEj9X9vBeToQ2rLCSJqb11Ue9UELs6Mwmrj9z3UqTDyunEF3oZGQRg6cI5rvhoC"
+    "9Qy6UIg+M9wi/42bSM8ESzrSD8sreBcmmNTiLLDi0UnJIb7skgJViU08Es88tYrqSBYrq7"
+    "ZUAdezK8NbpiVWbEpNTk3c9O8u3DwE/Mbnc6/SttAiEao4+t/kgfDfTRHVnGjz1e7KvqoV"
+    "+TlftYAx91+oJF1IW7heaFNtoj9Ai6u+NdQgPD9Xk7g0LDcoDNB24hw3LasdBpIlxL/ErA"
+    "uelcd3ut2xeXJ+dMuCfVyZdFQ9QNeTvUioOTCwFJxspDxyVD8MDBUzz/p/XV5L/55D+Fob"
+    "FZ2kReUoW46hXiUjHpZ9GvxZh0EhioGL7MS6kIZja+smUQk0kHry1+K2OZeR2p0+YRLYh9"
+    "zVmDE7ZLrpyUhevAsk5t9o0QH7y75vEZSQduF0urgzsiDqF9hX7A3d9AzElkROSke/dCrZ"
+    "FPowKISXU5AWyuxZc3S/jyJocvF23kW8JNCjfyPZh7srdZd2eOyFHXe/34P9+ncVw="
+)

pyproject.toml

@@ -32,6 +32,8 @@ dependencies = [
     "langchain-openai>=1.1.6",
     "langchain>=1.2.0",
     "langchain-community>=0.4.1",
+    "langgraph-checkpoint-sqlite>=2.0.0",
+    "aiosqlite>=0.20.0",
     "jq>=1.10.0",
     "tavily-python>=0.7.17",
     "ynab>=1.3.0",
@@ -40,6 +42,11 @@ dependencies = [
     "ollama>=0.6.1",
     "twilio>=9.10.2",
     "aioboto3>=13.0.0",
+    "prometheus-client>=0.26.0",
+    "google-api-python-client>=2.0.0",
+    "google-auth-oauthlib>=1.0.0",
+    "google-auth-httplib2>=0.2.0",
+    "langchain-mcp-adapters>=0.3.2",
 ]
 
 [project.optional-dependencies]

raggr-frontend/src/api/conversationService.ts

@@ -1,18 +1,20 @@
 import { userService } from "./userService";
 
 export type SSEEvent =
+  | { type: "thinking" }
+  | { type: "thinking_delta"; delta: string }
   | { type: "tool_start"; tool: string }
   | { type: "tool_end"; tool: string }
   | { type: "content"; delta: string }
   | { type: "response"; message: string }
-  | { type: "error"; message: string };
+  | { type: "error"; message: string; stalled?: boolean };
 
 export type SSEEventCallback = (event: SSEEvent) => void;
 
 interface Message {
   id: string;
   text: string;
-  speaker: "user" | "simba";
+  speaker: "user" | "simba" | "tool";
   created_at: string;
   image_key?: string | null;
 }

raggr-frontend/src/components/ChatScreen.tsx

@@ -1,9 +1,10 @@
-import { useCallback, useState, useRef } from "react";
+import React, { useCallback, useState, useRef } from "react";
 import { useNavigate } from "react-router-dom";
 import { PanelLeftOpen, Menu, X } from "lucide-react";
 import { QuestionBubble } from "./QuestionBubble";
 import { AnswerBubble } from "./AnswerBubble";
 import { ToolBubble } from "./ToolBubble";
+import { ThinkingBubble } from "./ThinkingBubble";
 import { MessageInput } from "./MessageInput";
 import { SidebarContent } from "./SidebarContent";
 import { AdminPanel } from "./AdminPanel";
@@ -53,6 +54,7 @@ export const ChatScreen = ({ setAuthenticated, isAdmin }: ChatScreenProps) => {
     messages,
     isLoading,
     streaming,
+    thinking,
     messagesLoading,
     pendingImage,
     setPendingImage,
@@ -60,6 +62,10 @@ export const ChatScreen = ({ setAuthenticated, isAdmin }: ChatScreenProps) => {
     stopGeneration,
     error,
     clearError,
+    stalled,
+    retryStalled,
+    thinkingText,
+    thinkingDurationMs,
   } = useChat({
     createConversation,
     refreshConversations,
@@ -242,14 +248,50 @@ export const ChatScreen = ({ setAuthenticated, isAdmin }: ChatScreenProps) => {
             <div className="flex-1 overflow-y-auto px-4 py-6">
               <div className="max-w-3xl mx-auto flex flex-col gap-6">
                 {messages.map((msg, index) => {
-                  if (msg.speaker === "tool")
-                    return <ToolBubble key={index} text={msg.text} />;
-                  if (msg.speaker === "simba")
-                    return <AnswerBubble key={index} text={msg.text} />;
-                  return <QuestionBubble key={index} text={msg.text} image_key={msg.image_key} />;
+                  // When the model has been thinking and content is now streaming,
+                  // render ThinkingBubble just before the streaming simba message
+                  // so it appears above the answer rather than below it.
+                  const showThinkingHere =
+                    streaming &&
+                    isLoading &&
+                    (thinking || thinkingText || thinkingDurationMs !== null) &&
+                    index === messages.length - 1 &&
+                    msg.speaker === "simba";
+
+                  const bubble =
+                    msg.speaker === "tool" ? (
+                      <ToolBubble key={index} text={msg.text} />
+                    ) : msg.speaker === "simba" ? (
+                      <AnswerBubble key={index} text={msg.text} />
+                    ) : (
+                      <QuestionBubble key={index} text={msg.text} image_key={msg.image_key} />
+                    );
+
+                  if (showThinkingHere) {
+                    return (
+                      <React.Fragment key={index}>
+                        <ThinkingBubble
+                          text={thinkingText}
+                          durationMs={thinkingDurationMs}
+                          active={thinking}
+                        />
+                        {bubble}
+                      </React.Fragment>
+                    );
+                  }
+                  return bubble;
                 })}
 
-                {((isLoading && !streaming) || messagesLoading) && (
+                {/* ThinkingBubble before content arrives (no streaming message yet) */}
+                {isLoading && !streaming && (thinking || thinkingText || thinkingDurationMs !== null) && (
+                  <ThinkingBubble
+                    text={thinkingText}
+                    durationMs={thinkingDurationMs}
+                    active={thinking}
+                  />
+                )}
+
+                {((isLoading && !streaming && !thinking) || messagesLoading) && (
                   <AnswerBubble text="" loading={true} />
                 )}
 
@@ -268,6 +310,20 @@ export const ChatScreen = ({ setAuthenticated, isAdmin }: ChatScreenProps) => {
                   </div>
                 )}
 
+                {stalled && (
+                  <div className="flex justify-center message-enter">
+                    <div className="inline-flex items-center gap-2 px-3 py-1.5 rounded-full bg-red-50 border border-red-200 text-xs text-red-600">
+                      <span>Simba didn&apos;t respond in time.</span>
+                      <button
+                        onClick={retryStalled}
+                        className="font-medium underline underline-offset-2 hover:text-red-800 transition-colors cursor-pointer"
+                      >
+                        Retry
+                      </button>
+                    </div>
+                  </div>
+                )}
+
                 <div ref={messagesEndRef} />
               </div>
             </div>

raggr-frontend/src/components/ThinkingBubble.tsx

@@ -0,0 +1,78 @@
+import React, { useState, useEffect, useRef } from "react";
+import { ChevronDown } from "lucide-react";
+import { cn } from "../lib/utils";
+
+type ThinkingBubbleProps = {
+  text: string;
+  durationMs: number | null;
+  active: boolean;
+};
+
+export const ThinkingBubble = React.memo(
+  ({ text, durationMs, active }: ThinkingBubbleProps) => {
+    const [expanded, setExpanded] = useState(true);
+    const textRef = useRef<HTMLDivElement>(null);
+
+    // Auto-scroll thinking text as it streams in.
+    useEffect(() => {
+      if (active && textRef.current) {
+        textRef.current.scrollTop = textRef.current.scrollHeight;
+      }
+    }, [text, active]);
+
+    // Collapse automatically when thinking finishes.
+    useEffect(() => {
+      if (durationMs !== null) {
+        setExpanded(false);
+      }
+    }, [durationMs]);
+
+    const label = active
+      ? "Thinking…"
+      : durationMs !== null
+        ? `Thought for ${(durationMs / 1000).toFixed(1)}s`
+        : "Thought";
+
+    return (
+      <div className="flex justify-center message-enter">
+        <div className="w-full max-w-2xl">
+          <button
+            onClick={() => setExpanded((v) => !v)}
+            className={cn(
+              "w-full flex items-center justify-between gap-2",
+              "px-3 py-1.5 rounded-xl",
+              "bg-leaf-pale border border-leaf-light/50",
+              "text-xs text-leaf-dark italic",
+              "hover:bg-leaf-light/20 transition-colors cursor-pointer",
+              active && "animate-pulse",
+            )}
+          >
+            <span>{label}</span>
+            {text && (
+              <ChevronDown
+                size={13}
+                className={cn(
+                  "text-leaf-dark/60 transition-transform duration-200 shrink-0",
+                  expanded && "rotate-180",
+                )}
+              />
+            )}
+          </button>
+
+          {expanded && text && (
+            <div
+              ref={textRef}
+              className={cn(
+                "mt-1 px-3 py-2 rounded-xl max-h-48 overflow-y-auto",
+                "bg-leaf-pale/40 border border-leaf-light/30",
+                "text-xs text-leaf-dark/70 font-mono leading-relaxed whitespace-pre-wrap",
+              )}
+            >
+              {text}
+            </div>
+          )}
+        </div>
+      </div>
+    );
+  },
+);

raggr-frontend/src/components/ToolBubble.tsx

@@ -1,16 +1,19 @@
 import React from "react";
 import { cn } from "../lib/utils";
 
-export const ToolBubble = React.memo(({ text }: { text: string }) => (
-  <div className="flex justify-center message-enter">
-    <div
-      className={cn(
-        "inline-flex items-center gap-1.5 px-3 py-1 rounded-full",
-        "bg-leaf-pale border border-leaf-light/50",
-        "text-xs text-leaf-dark italic",
-      )}
-    >
-      {text}
+export const ToolBubble = React.memo(
+  ({ text, pulse = false }: { text: string; pulse?: boolean }) => (
+    <div className="flex justify-center message-enter">
+      <div
+        className={cn(
+          "inline-flex items-center gap-1.5 px-3 py-1 rounded-full",
+          "bg-leaf-pale border border-leaf-light/50",
+          "text-xs text-leaf-dark italic",
+          pulse && "animate-pulse",
+        )}
+      >
+        {text}
+      </div>
     </div>
-  </div>
-));
+  ),
+);

raggr-frontend/src/hooks/useChat.ts

@@ -51,7 +51,14 @@ export function useChat({
   const [messagesLoading, setMessagesLoading] = useState(false);
   const [pendingImage, setPendingImage] = useState<File | null>(null);
   const [error, setError] = useState<string | null>(null);
+  const [stalled, setStalled] = useState(false);
   const [streaming, setStreaming] = useState(false);
+  const [thinking, setThinking] = useState(false);
+  const [thinkingText, setThinkingText] = useState("");
+  const [thinkingDurationMs, setThinkingDurationMs] = useState<number | null>(null);
+
+  const stalledQueryRef = useRef<string | null>(null);
+  const thinkingStartRef = useRef<number | null>(null);
 
   const isMountedRef = useRef(true);
   // True once tokens for the current assistant reply have started arriving, so
@@ -66,7 +73,9 @@ export function useChat({
     isMountedRef.current = true;
     return () => {
       isMountedRef.current = false;
-      abortControllerRef.current?.abort();
+      // Intentionally do NOT abort on unmount — navigation away lets the
+      // stream finish in the background so the response gets persisted.
+      // stopGeneration() still aborts on explicit user cancel.
     };
   }, []);
 
@@ -141,6 +150,11 @@ export function useChat({
       const imageFile = pendingImage;
 
       setError(null);
+      setStalled(false);
+      stalledQueryRef.current = null;
+      setThinkingText("");
+      setThinkingDurationMs(null);
+      thinkingStartRef.current = null;
       streamingActiveRef.current = false;
       setStreaming(false);
       updateMessages((prev) => prev.concat([{ text: query, speaker: "user" }]));
@@ -186,15 +200,27 @@ export function useChat({
           activeId,
           (event) => {
             if (!isMountedRef.current) return;
-            if (event.type === "tool_start") {
+            if (event.type === "thinking") {
+              setThinking(true);
+              if (thinkingStartRef.current === null) thinkingStartRef.current = Date.now();
+            } else if (event.type === "thinking_delta") {
+              setThinking(true);
+              if (thinkingStartRef.current === null) thinkingStartRef.current = Date.now();
+              setThinkingText((prev) => prev + event.delta);
+            } else if (event.type === "tool_start") {
               // Any text after a tool call belongs in a fresh bubble.
               streamingActiveRef.current = false;
+              setThinking(false);
               const friendly =
                 TOOL_MESSAGES[event.tool] ?? `Using ${event.tool}...`;
               updateMessages((prev) =>
                 prev.concat([{ text: friendly, speaker: "tool" }]),
               );
             } else if (event.type === "content") {
+              setThinking(false);
+              if (thinkingStartRef.current !== null && thinkingDurationMs === null) {
+                setThinkingDurationMs(Date.now() - thinkingStartRef.current);
+              }
               if (streamingActiveRef.current) {
                 updateMessages((prev) => {
                   const updated = [...prev];
@@ -232,7 +258,11 @@ export function useChat({
               }
             } else if (event.type === "error") {
               console.error("Stream error:", event.message);
-              if (isMountedRef.current) {
+              if (!isMountedRef.current) return;
+              if (event.stalled) {
+                stalledQueryRef.current = query;
+                setStalled(true);
+              } else {
                 setError("Simba ran into a problem answering that.");
               }
             }
@@ -258,6 +288,7 @@ export function useChat({
         if (isMountedRef.current) {
           setIsLoading(false);
           setStreaming(false);
+          setThinking(false);
           // Refresh after every send so the just-used conversation floats to
           // the top of the sidebar (sorted by last activity) and picks up the
           // auto-generated title on the first message.
@@ -285,11 +316,20 @@ export function useChat({
 
   const clearError = useCallback(() => setError(null), []);
 
+  const retryStalled = useCallback(() => {
+    const q = stalledQueryRef.current;
+    if (!q) return;
+    setStalled(false);
+    stalledQueryRef.current = null;
+    sendMessage(q, false);
+  }, [sendMessage]);
+
   return {
     messages,
     setMessages: updateMessages,
     isLoading,
     streaming,
+    thinking,
     messagesLoading,
     pendingImage,
     setPendingImage,
@@ -297,5 +337,9 @@ export function useChat({
     stopGeneration,
     error,
     clearError,
+    stalled,
+    retryStalled,
+    thinkingText,
+    thinkingDurationMs,
   };
 }

scripts/google_auth.py

@@ -0,0 +1,62 @@
+# /// script
+# requires-python = ">=3.13"
+# dependencies = [
+#   "google-auth-oauthlib>=1.0.0",
+# ]
+# ///
+"""One-time OAuth2 setup to generate token.json for Google Calendar and Gmail.
+
+Usage:
+    uv run scripts/google_auth.py [--credentials credentials.json] [--token token.json]
+
+Requires credentials.json downloaded from Google Cloud Console
+(APIs & Services → Credentials → OAuth 2.0 Client ID → Desktop app).
+"""
+
+import argparse
+import sys
+from pathlib import Path
+
+from google_auth_oauthlib.flow import InstalledAppFlow
+
+SCOPES = [
+    "https://www.googleapis.com/auth/calendar.events",
+    "https://www.googleapis.com/auth/gmail.readonly",
+]
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Authorize Google APIs for raggr")
+    parser.add_argument(
+        "--credentials",
+        default="credentials.json",
+        help="Path to OAuth2 credentials JSON from Google Cloud Console",
+    )
+    parser.add_argument(
+        "--token",
+        default="token.json",
+        help="Output path for the generated token file",
+    )
+    args = parser.parse_args()
+
+    creds_path = Path(args.credentials)
+    if not creds_path.exists():
+        print(f"Error: credentials file not found: {creds_path}")
+        print(
+            "Download it from Google Cloud Console → APIs & Services → Credentials → OAuth 2.0 Client ID"
+        )
+        sys.exit(1)
+
+    flow = InstalledAppFlow.from_client_secrets_file(str(creds_path), SCOPES)
+    creds = flow.run_local_server(port=0)
+
+    token_path = Path(args.token)
+    token_path.write_text(creds.to_json())
+    print(f"Token saved to {token_path}")
+    print(
+        f"Mount this file into Docker: - ./{token_path}:/app/config/google-token.json:ro"
+    )
+
+
+if __name__ == "__main__":
+    main()

scripts/manage_beli.py

@@ -0,0 +1,238 @@
+"""
+Manage Beli profiles and restaurant data.
+
+Usage:
+  uv run python scripts/manage_beli.py add <username> <beli_email> <beli_password>
+  uv run python scripts/manage_beli.py list
+  uv run python scripts/manage_beli.py refresh <profile_id>
+  uv run python scripts/manage_beli.py refresh-all
+  uv run python scripts/manage_beli.py remove <profile_id>
+
+Examples:
+  uv run python scripts/manage_beli.py add ryan me@gmail.com mybelipass
+  uv run python scripts/manage_beli.py list
+  uv run python scripts/manage_beli.py refresh 3f2a1b...
+  uv run python scripts/manage_beli.py refresh-all
+"""
+
+import asyncio
+import os
+import sys
+
+from dotenv import load_dotenv
+from tortoise import Tortoise
+
+load_dotenv()
+
+DATABASE_URL = os.getenv(
+    "DATABASE_URL", "postgres://raggr:changeme@localhost:5432/raggr"
+)
+
+TORTOISE_CONFIG = {
+    "connections": {"default": DATABASE_URL},
+    "apps": {
+        "models": {
+            "models": [
+                "blueprints.users.models",
+                "blueprints.beli.models",
+                "blueprints.email.models",
+                "blueprints.conversation.models",
+                "blueprints.scheduled_messages.models",
+                "blueprints.cron.models",
+            ],
+            "default_connection": "default",
+        }
+    },
+}
+
+
+async def _init():
+    await Tortoise.init(config=TORTOISE_CONFIG)
+
+
+async def _close():
+    await Tortoise.close_connections()
+
+
+async def cmd_add(username: str, beli_email: str, beli_password: str):
+    from blueprints.beli.logic import MAX_PROFILES_PER_USER
+    from blueprints.beli.models import BeliProfile
+    from blueprints.users.models import User
+    from utils.beli_service import authenticate
+
+    user = await User.get_or_none(username=username)
+    if not user:
+        print(f"Error: no user with username '{username}'")
+        return False
+
+    count = await BeliProfile.filter(user=user).count()
+    if count >= MAX_PROFILES_PER_USER:
+        print(
+            f"Error: user already has {MAX_PROFILES_PER_USER} Beli profiles (maximum)"
+        )
+        return False
+
+    if await BeliProfile.exists(user=user, beli_email=beli_email):
+        print(f"Error: profile for '{beli_email}' already exists for this user")
+        return False
+
+    print(f"Authenticating with Beli as {beli_email}...")
+    try:
+        session = await authenticate(beli_email, beli_password)
+    except Exception as e:
+        print(f"Error: Beli authentication failed — {e}")
+        return False
+
+    profile = await BeliProfile.create(
+        user=user,
+        beli_email=beli_email,
+        beli_password=beli_password,
+        beli_user_id=session["user_id"],
+        beli_refresh_token=session["refresh_token"],
+        enabled=True,
+    )
+    print(f"Profile created: {profile.id}")
+    print(f"  Beli user ID : {profile.beli_user_id}")
+    print(f"  Beli email   : {profile.beli_email}")
+    print(f"  Owner        : {username}")
+    return True
+
+
+async def cmd_list():
+    from blueprints.beli.models import BeliProfile, BeliRestaurant
+
+    profiles = await BeliProfile.all().prefetch_related("user").order_by("created_at")
+    if not profiles:
+        print("No Beli profiles found.")
+        return
+
+    for p in profiles:
+        count = await BeliRestaurant.filter(profile=p).count()
+        refreshed = (
+            p.last_refreshed_at.strftime("%Y-%m-%d %H:%M UTC")
+            if p.last_refreshed_at
+            else "never"
+        )
+        status = "enabled" if p.enabled else "DISABLED"
+        print(f"{p.id}  [{status}]")
+        print(f"  Owner    : {p.user.email}")
+        print(f"  Beli     : {p.beli_email}  (user_id={p.beli_user_id})")
+        print(f"  Restaurants: {count}  |  Last refresh: {refreshed}")
+        if p.last_error:
+            print(f"  Last error : {p.last_error}")
+        print()
+
+
+async def cmd_refresh(profile_id: str):
+    from blueprints.beli.logic import refresh_profile
+    from blueprints.beli.models import BeliProfile
+
+    profile = await BeliProfile.get_or_none(id=profile_id)
+    if not profile:
+        print(f"Error: no profile with id '{profile_id}'")
+        return False
+
+    print(f"Refreshing profile {profile_id} ({profile.beli_email})...")
+    result = await refresh_profile(profile_id)
+    if result.get("ok"):
+        print(f"Done — {result['upserted']} restaurants upserted")
+    elif result.get("skipped"):
+        print("Skipped (profile disabled)")
+    else:
+        print(f"Failed: {result.get('error')}")
+        return False
+    return True
+
+
+async def cmd_refresh_all():
+    from blueprints.beli.models import BeliProfile
+    from blueprints.beli.logic import refresh_profile
+
+    profiles = (
+        await BeliProfile.filter(enabled=True).order_by("last_refreshed_at").all()
+    )
+    if not profiles:
+        print("No enabled Beli profiles found.")
+        return
+
+    print(f"Refreshing {len(profiles)} profile(s)...")
+    for p in profiles:
+        print(f"\n  {p.beli_email} ({p.id})")
+        result = await refresh_profile(str(p.id))
+        if result.get("ok"):
+            print(f"  → {result['upserted']} restaurants upserted")
+        else:
+            print(f"  → failed: {result.get('error')}")
+
+
+async def cmd_remove(profile_id: str):
+    from blueprints.beli.models import BeliProfile, BeliRestaurant
+
+    profile = await BeliProfile.get_or_none(id=profile_id)
+    if not profile:
+        print(f"Error: no profile with id '{profile_id}'")
+        return False
+
+    deleted = await BeliRestaurant.filter(profile=profile).count()
+    await BeliRestaurant.filter(profile=profile).delete()
+    await profile.delete()
+    print(
+        f"Removed profile {profile_id} ({profile.beli_email}) and {deleted} restaurants"
+    )
+    return True
+
+
+def usage():
+    print(__doc__)
+
+
+async def main():
+    if len(sys.argv) < 2:
+        usage()
+        sys.exit(1)
+
+    cmd = sys.argv[1]
+
+    await _init()
+    try:
+        if cmd == "add":
+            if len(sys.argv) != 5:
+                print(
+                    "Usage: uv run python scripts/manage_beli.py add <username> <beli_email> <beli_password>"
+                )
+                sys.exit(1)
+            ok = await cmd_add(sys.argv[2], sys.argv[3], sys.argv[4])
+            sys.exit(0 if ok else 1)
+
+        elif cmd == "list":
+            await cmd_list()
+
+        elif cmd == "refresh":
+            if len(sys.argv) != 3:
+                print(
+                    "Usage: uv run python scripts/manage_beli.py refresh <profile_id>"
+                )
+                sys.exit(1)
+            ok = await cmd_refresh(sys.argv[2])
+            sys.exit(0 if ok else 1)
+
+        elif cmd == "refresh-all":
+            await cmd_refresh_all()
+
+        elif cmd == "remove":
+            if len(sys.argv) != 3:
+                print("Usage: uv run python scripts/manage_beli.py remove <profile_id>")
+                sys.exit(1)
+            ok = await cmd_remove(sys.argv[2])
+            sys.exit(0 if ok else 1)
+
+        else:
+            print(f"Unknown command: {cmd}")
+            usage()
+            sys.exit(1)
+    finally:
+        await _close()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

scripts/migrate-new.sh

@@ -0,0 +1,43 @@
+#!/bin/bash
+# Generate a new aerich migration.
+# If a raggr container is running, exec into it.
+# Otherwise, spin up the local postgres service and run aerich from the venv.
+set -euo pipefail
+
+cd "$(dirname "$0")/.."
+
+# Load .env so POSTGRES_* vars are available locally
+if [ -f .env ]; then
+    set -a
+    # shellcheck disable=SC1091
+    source .env
+    set +a
+fi
+
+POSTGRES_USER="${POSTGRES_USER:-raggr}"
+POSTGRES_PASSWORD="${POSTGRES_PASSWORD:-changeme}"
+POSTGRES_DB="${POSTGRES_DB:-raggr}"
+LOCAL_DB_URL="postgres://${POSTGRES_USER}:${POSTGRES_PASSWORD}@localhost:5432/${POSTGRES_DB}"
+
+read -rp "Migration name: " name
+
+RUNNING=$(docker compose ps --status running --services 2>/dev/null | grep -oE 'raggr-(blue|green)' | head -1 || true)
+
+if [ -n "$RUNNING" ]; then
+    echo "==> Using running container: $RUNNING"
+    docker compose --profile blue --profile green exec "$RUNNING" aerich migrate --name "$name"
+else
+    echo "==> No raggr container running — using local postgres"
+    docker compose up -d postgres
+
+    echo "==> Waiting for postgres to be ready..."
+    until docker compose exec -T postgres pg_isready -U "$POSTGRES_USER" > /dev/null 2>&1; do
+        sleep 1
+    done
+
+    echo "==> Applying existing migrations..."
+    DATABASE_URL="$LOCAL_DB_URL" .venv/bin/aerich upgrade
+
+    echo "==> Generating migration: $name"
+    DATABASE_URL="$LOCAL_DB_URL" .venv/bin/aerich migrate --name "$name"
+fi

tests/unit/test_homebox_mcp.py

@@ -0,0 +1,122 @@
+"""Tests for the admin-only Homebox MCP tool wrapping and agent routing."""
+
+import os
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from langchain_core.tools import StructuredTool
+from quart import Quart
+
+from utils.homebox_mcp import (
+    ADMIN_ONLY_MESSAGE,
+    HOMEBOX_TOOL_ALLOWLIST,
+    load_homebox_tools,
+    select_homebox_tools,
+)
+
+ADMIN = {"configurable": {"is_admin": True}}
+NON_ADMIN = {"configurable": {"is_admin": False}}
+
+
+def _mcp_tool(name: str, result=None, error: Exception | None = None):
+    """Build a stub shaped like a langchain-mcp-adapters tool."""
+    calls = []
+
+    async def call(runtime=None, **arguments):
+        calls.append(arguments)
+        if error:
+            raise error
+        return (
+            result if result is not None else ([{"type": "text", "text": "ok"}], None)
+        )
+
+    tool = StructuredTool(
+        name=name,
+        description=f"{name} description",
+        args_schema={
+            "type": "object",
+            "properties": {"query": {"type": "string"}},
+        },
+        coroutine=call,
+        response_format="content_and_artifact",
+    )
+    return tool, calls
+
+
+class TestSelectHomeboxTools:
+    def test_filters_to_allowlist_and_prefixes(self):
+        stubs = [_mcp_tool(n)[0] for n in ("search_items", "delete_item", "get_item")]
+        tools = select_homebox_tools(stubs)
+        assert [t.name for t in tools] == ["homebox_search_items", "homebox_get_item"]
+
+    def test_never_exposes_deletes(self):
+        assert not any(name.startswith("delete_") for name in HOMEBOX_TOOL_ALLOWLIST)
+
+    def test_truncates_long_descriptions(self):
+        stub, _ = _mcp_tool("search_items")
+        stub.description = "word " * 200
+        [tool] = select_homebox_tools([stub])
+        assert len(tool.description) <= 300
+
+
+class TestAdminGuard:
+    async def test_admin_call_delegates_and_flattens_content(self):
+        stub, calls = _mcp_tool(
+            "search_items",
+            result=(
+                [{"type": "text", "text": "Drill"}, {"type": "text", "text": "Garage"}],
+                None,
+            ),
+        )
+        [tool] = select_homebox_tools([stub])
+        out = await tool.ainvoke({"query": "drill"}, config=ADMIN)
+        assert out == "Drill\nGarage"
+        assert calls == [{"query": "drill"}]
+
+    async def test_non_admin_is_refused_without_calling_server(self):
+        stub, calls = _mcp_tool("search_items")
+        [tool] = select_homebox_tools([stub])
+        assert (
+            await tool.ainvoke({"query": "drill"}, config=NON_ADMIN)
+            == ADMIN_ONLY_MESSAGE
+        )
+        assert await tool.ainvoke({"query": "drill"}) == ADMIN_ONLY_MESSAGE
+        assert calls == []
+
+    async def test_server_errors_are_returned_as_text(self):
+        stub, _ = _mcp_tool("search_items", error=RuntimeError("cannot reach Homebox"))
+        [tool] = select_homebox_tools([stub])
+        out = await tool.ainvoke({"query": "drill"}, config=ADMIN)
+        assert out == "Homebox error: cannot reach Homebox"
+
+    def test_config_not_exposed_in_schema(self):
+        [tool] = select_homebox_tools([_mcp_tool("search_items")[0]])
+        assert list(tool.tool_call_schema["properties"]) == ["query"]
+
+
+class TestLoadHomeboxTools:
+    async def test_returns_empty_when_unconfigured(self):
+        with patch.dict(os.environ, {"HOMEBOX_URL": "", "HOMEBOX_TOKEN": ""}):
+            assert await load_homebox_tools() == []
+
+    async def test_returns_empty_when_server_fails(self):
+        env = {
+            "HOMEBOX_URL": "http://homebox",
+            "HOMEBOX_TOKEN": "hb_x",
+            "HOMEBOX_MCP_COMMAND": "/nonexistent/homebox-mcp",
+        }
+        with patch.dict(os.environ, env):
+            assert await load_homebox_tools() == []
+
+
+class TestGetAgentFor:
+    async def test_routes_admin_to_admin_agent(self):
+        from blueprints.conversation.logic import get_agent_for
+
+        app = Quart(__name__)
+        app.main_agent, app.admin_agent = "main", "admin"
+        admin = SimpleNamespace(is_admin=lambda: True)
+        user = SimpleNamespace(is_admin=lambda: False)
+        async with app.app_context():
+            assert get_agent_for(admin) == "admin"
+            assert get_agent_for(user) == "main"

tests/unit/test_metrics.py

@@ -0,0 +1,109 @@
+"""Tests for the /metrics endpoint (token gate) and the metrics helpers."""
+
+import os
+import time
+from unittest.mock import patch
+
+import pytest
+from quart import Quart
+
+import app as app_module
+from metrics import CHAT_REQUESTS_TOTAL, render_metrics, time_response
+
+
+def _make_app() -> Quart:
+    """Build a minimal Quart app exposing only the production /metrics handler.
+
+    Reuses the real handler from ``app.py`` but registers it on a fresh app
+    without the full app's ``@app.while_serving`` lifespan, so no Tortoise DB
+    connection is attempted and the route can be tested hermetically.
+    """
+    minimal = Quart(__name__)
+    minimal.add_url_rule(
+        "/metrics",
+        view_func=app_module.app.view_functions["metrics_endpoint"],
+        methods=["GET"],
+    )
+    return minimal
+
+
+def _counter_value(stream: str, result: str) -> float:
+    return CHAT_REQUESTS_TOTAL.labels(stream=stream, result=result)._value.get()
+
+
+class TestMetricsEndpoint:
+    async def test_403_when_token_unset(self):
+        app = _make_app()
+        with patch.dict(os.environ, {}, clear=True):
+            async with app.test_client() as client:
+                resp = await client.get("/metrics")
+        assert resp.status_code == 403
+
+    async def test_403_when_token_empty(self):
+        app = _make_app()
+        with patch.dict(os.environ, {"METRICS_TOKEN": ""}):
+            async with app.test_client() as client:
+                resp = await client.get("/metrics")
+        assert resp.status_code == 403
+
+    async def test_403_wrong_token(self):
+        app = _make_app()
+        with patch.dict(os.environ, {"METRICS_TOKEN": "expected-secret"}):
+            async with app.test_client() as client:
+                resp = await client.get(
+                    "/metrics",
+                    headers={"Authorization": "Bearer wrong-secret"},
+                )
+        assert resp.status_code == 403
+
+    async def test_403_missing_auth_header(self):
+        app = _make_app()
+        with patch.dict(os.environ, {"METRICS_TOKEN": "expected-secret"}):
+            async with app.test_client() as client:
+                resp = await client.get("/metrics")
+        assert resp.status_code == 403
+
+    async def test_200_correct_token_and_metric_present(self):
+        app = _make_app()
+        with patch.dict(os.environ, {"METRICS_TOKEN": "expected-secret"}):
+            with time_response(stream=False):
+                time.sleep(0.001)
+            async with app.test_client() as client:
+                resp = await client.get(
+                    "/metrics",
+                    headers={"Authorization": "Bearer expected-secret"},
+                )
+        assert resp.status_code == 200
+        assert "text/plain" in resp.content_type
+        body = await resp.get_data(as_text=True)
+        assert "raggr_chat_response_seconds" in body
+
+
+class TestTimeResponse:
+    def test_counts_success(self):
+        before = _counter_value("false", "success")
+        with time_response(stream=False):
+            pass
+        after = _counter_value("false", "success")
+        assert after == before + 1
+
+    def test_counts_error_on_exception(self):
+        before = _counter_value("false", "error")
+        with pytest.raises(RuntimeError):
+            with time_response(stream=False):
+                raise RuntimeError("boom")
+        after = _counter_value("false", "error")
+        assert after == before + 1
+
+    def test_observes_histogram_value(self):
+        with time_response(stream=False):
+            time.sleep(0.001)
+        out = render_metrics().decode()
+        assert "raggr_chat_response_seconds" in out
+
+    def test_stream_labels_distinct(self):
+        before = _counter_value("true", "success")
+        with time_response(stream=True):
+            pass
+        after = _counter_value("true", "success")
+        assert after == before + 1

utils/beli_service.py

@@ -0,0 +1,182 @@
+"""
+Beli API client.
+
+Reverse-engineered from ProjectBarks/beli-api (mirrored at git.torrtle.co/ryan/beli-api).
+Hosts and auth flow discovered from observed traffic — may break without notice.
+
+Auth: POST /api/token/ → {access, refresh}  (access: 20 min, refresh: 7 days)
+The API rejects requests without an Origin header (app uses capacitor://localhost;
+https://localhost also works per external clients).
+"""
+
+import base64
+import json
+import logging
+import random
+import time
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+HOST_ONBOARD = "https://backoffice-service-onboarding-t57o3dxfca-nn.a.run.app"
+HOST_API = "https://backoffice-service-t57o3dxfca-nn.a.run.app"
+
+# 3-letter category codes used by the Beli API
+RANKING_CATEGORIES = ["RES", "DES", "BAR", "BAK"]
+
+# 350 ms between requests — community-polite interval per API reference
+REQUEST_SPACING = 0.35
+
+_USER_AGENTS = [
+    "Mozilla/5.0 (iPhone; CPU iPhone OS 18_7 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148",
+    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
+    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
+    "Mozilla/5.0 (Linux; Android 16; SM-S928U) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/149.0.7827.91 Mobile Safari/537.36",
+    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/18.1 Safari/605.1.15",
+]
+
+
+def _pick_ua() -> str:
+    return random.choice(_USER_AGENTS)
+
+
+def _base_headers(access_token: str | None = None, ua: str | None = None) -> dict:
+    h = {
+        "User-Agent": ua or _pick_ua(),
+        "Origin": "https://localhost",
+        "Content-Type": "application/json",
+    }
+    if access_token:
+        h["Authorization"] = f"Bearer {access_token}"
+    return h
+
+
+def _decode_jwt_claims(token: str) -> dict:
+    payload = token.split(".")[1]
+    payload += "=" * (-len(payload) % 4)
+    return json.loads(base64.urlsafe_b64decode(payload))
+
+
+def is_token_expired(token: str | None, skew_seconds: int = 60) -> bool:
+    if not token:
+        return True
+    try:
+        return _decode_jwt_claims(token)["exp"] - skew_seconds <= time.time()
+    except Exception:
+        return True
+
+
+async def authenticate(email: str, password: str) -> dict:
+    """Login with email+password. Returns {access_token, refresh_token, user_id}."""
+    ua = _pick_ua()
+    async with httpx.AsyncClient() as client:
+        resp = await client.post(
+            f"{HOST_ONBOARD}/api/token/",
+            json={"email": email, "password": password},
+            headers=_base_headers(ua=ua),
+            timeout=15,
+        )
+        resp.raise_for_status()
+        data = resp.json()
+
+    claims = _decode_jwt_claims(data["access"])
+    return {
+        "access_token": data["access"],
+        "refresh_token": data["refresh"],
+        "user_id": claims["user_id"],
+    }
+
+
+async def refresh_access_token(refresh_token: str) -> str:
+    """Exchange a refresh token for a new access token."""
+    async with httpx.AsyncClient() as client:
+        resp = await client.post(
+            f"{HOST_ONBOARD}/api/token/refresh/",
+            json={"refresh": refresh_token},
+            headers=_base_headers(),
+            timeout=15,
+        )
+        resp.raise_for_status()
+        return resp.json()["access"]
+
+
+async def fetch_user_rankings(
+    access_token: str, user_id: str, category: str
+) -> list[dict]:
+    """Fetch a user's ranked places for one category (RES/DES/BAR/BAK)."""
+    async with httpx.AsyncClient() as client:
+        resp = await client.get(
+            f"{HOST_API}/api/get-ranking/",
+            params={"user": user_id, "category": category},
+            headers=_base_headers(access_token),
+            timeout=30,
+        )
+        resp.raise_for_status()
+        return resp.json().get("results") or []
+
+
+async def fetch_following(access_token: str, user_id: str) -> list[dict]:
+    """Fetch users that user_id follows. Returns list of user objects with id, full_name, username."""
+    async with httpx.AsyncClient() as client:
+        resp = await client.get(
+            f"{HOST_API}/api/following/{user_id}/",
+            headers=_base_headers(access_token),
+            timeout=30,
+        )
+        resp.raise_for_status()
+        data = resp.json()
+        # Endpoint returns either a paginated {results:[]} or bare list
+        if isinstance(data, list):
+            return data
+        return data.get("results") or []
+
+
+async def fetch_bookmarks(access_token: str) -> list[dict]:
+    """Fetch the authenticated user's bookmarked (want-to-try) places.
+
+    The endpoint takes no query params — it returns bookmarks for whoever
+    owns the bearer token.
+    """
+    async with httpx.AsyncClient() as client:
+        resp = await client.get(
+            f"{HOST_API}/api/get-bookmark/",
+            headers=_base_headers(access_token),
+            timeout=30,
+        )
+        resp.raise_for_status()
+        return resp.json().get("results") or []
+
+
+def extract_business(item: dict, rank_index: int | None = None) -> dict | None:
+    """Pull place fields from a ranking or bookmark result item.
+
+    Ranking items have a nested 'business' object (integer id) plus top-level
+    score fields. The score is stored as 'value' (sentiment seed sent by client;
+    server computes the displayed 0-10 score server-side).
+    Rank is the position in the results list, not a field in the response.
+    Notes are stored separately via data-user-business-text; we don't fetch them.
+    """
+    biz = item.get("business") or item
+    biz_id = biz.get("id") or item.get("business_id")
+    if not biz_id:
+        return None
+
+    price_key = biz.get("price_key")
+    if not price_key:
+        price_int = biz.get("price")
+        if price_int:
+            price_key = "$" * price_int
+
+    return {
+        "beli_business_id": str(biz_id),
+        "name": biz.get("name", ""),
+        "city": biz.get("city"),
+        "neighborhood": biz.get("neighborhood") or biz.get("borough"),
+        "cuisine": biz.get("cuisines") or [],
+        "price_range": price_key,
+        "rank_index": rank_index,
+        "score": item.get("value"),
+        "notes": None,
+        "tags": item.get("tags") or [],
+    }

utils/cronicle.py

@@ -0,0 +1,233 @@
+import logging
+import os
+from datetime import datetime, timedelta, timezone
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+
+def _base_url() -> str:
+    return os.environ.get("CRONICLE_URL", "").rstrip("/")
+
+
+def _api_key() -> str:
+    return os.environ.get("CRONICLE_API_KEY", "")
+
+
+def _is_configured() -> bool:
+    return bool(_base_url() and _api_key())
+
+
+async def _get_first_category_id() -> str:
+    async with httpx.AsyncClient() as client:
+        resp = await client.get(
+            f"{_base_url()}/api/app/get_categories/v1",
+            params={"api_key": _api_key()},
+            timeout=10,
+        )
+        resp.raise_for_status()
+        data = resp.json()
+        if data.get("code") != 0:
+            raise RuntimeError(f"Cronicle get_categories error: {data}")
+        rows = data.get("rows", [])
+        if not rows:
+            raise RuntimeError("Cronicle has no categories configured")
+        return rows[0]["id"]
+
+
+def _make_trigger_script() -> str:
+    """Node.js script that tries raggr-green then falls back to raggr-blue."""
+    cron_secret = os.environ.get("CRON_SECRET", "")
+    path = "/api/scheduled-messages/trigger"
+    return f"""#!/usr/bin/env node
+const http = require('http');
+function post(host) {{
+  return new Promise((resolve, reject) => {{
+    const req = http.request(
+      {{ hostname: host, port: 8080, path: '{path}', method: 'POST',
+         headers: {{'X-Cron-Secret': '{cron_secret}'}} }},
+      res => {{ res.resume(); resolve(res.statusCode); }}
+    );
+    req.setTimeout(300000, () => {{ req.destroy(); reject(new Error('timeout')); }});
+    req.on('error', reject);
+    req.end();
+  }});
+}}
+post('raggr-green').catch(() => post('raggr-blue')).catch(err => {{
+  console.error('trigger failed:', err.message);
+  process.exitCode = 1;
+}});
+"""
+
+
+def _build_timing(scheduled_at: datetime, recurrence: str) -> dict:
+    """Build a Cronicle timing object from a scheduled datetime and recurrence.
+
+    - none:    one-shot (fires exactly once at the given UTC datetime)
+    - daily:   fires every day at the same hour/minute
+    - weekly:  fires every week on the same weekday and hour/minute
+    - monthly: fires every month on the same day-of-month and hour/minute
+    """
+    utc = scheduled_at.astimezone(timezone.utc)
+    # Round up to the next minute if there are sub-minute seconds so Cronicle
+    # fires after the message is due, not before.
+    if utc.second or utc.microsecond:
+        utc = (utc + timedelta(minutes=1)).replace(second=0, microsecond=0)
+    h, m = utc.hour, utc.minute
+
+    if recurrence == "daily":
+        return {"hours": [h], "minutes": [m]}
+    elif recurrence == "weekly":
+        # weekday: 0=Sunday in Cronicle (Python isoweekday: Mon=1..Sun=7)
+        wd = utc.isoweekday() % 7  # convert to 0=Sunday
+        return {"weekdays": [wd], "hours": [h], "minutes": [m]}
+    elif recurrence == "monthly":
+        return {"days": [utc.day], "hours": [h], "minutes": [m]}
+    else:
+        # one-shot
+        return {
+            "years": [utc.year],
+            "months": [utc.month],
+            "days": [utc.day],
+            "hours": [utc.hour],
+            "minutes": [utc.minute],
+        }
+
+
+def _make_beli_refresh_script() -> str:
+    """Node.js script that triggers the Beli nightly refresh endpoint."""
+    cron_secret = os.environ.get("CRON_SECRET", "")
+    path = "/api/beli/cron/refresh"
+    return f"""#!/usr/bin/env node
+const http = require('http');
+function post(host) {{
+  return new Promise((resolve, reject) => {{
+    const req = http.request(
+      {{ hostname: host, port: 8080, path: '{path}', method: 'POST',
+         headers: {{'X-Cron-Secret': '{cron_secret}'}} }},
+      res => {{ res.resume(); resolve(res.statusCode); }}
+    );
+    req.setTimeout(30000, () => {{ req.destroy(); reject(new Error('timeout')); }});
+    req.on('error', reject);
+    req.end();
+  }});
+}}
+post('raggr-green').catch(() => post('raggr-blue')).catch(err => {{
+  console.error('beli refresh trigger failed:', err.message);
+  process.exitCode = 1;
+}});
+"""
+
+
+async def ensure_beli_nightly_event() -> str | None:
+    """Create (once) a daily 1 AM Cronicle event that triggers Beli nightly refresh.
+
+    Safe to call on every startup — checks for an existing event by title first.
+    Returns the Cronicle event ID, or None if Cronicle is not configured.
+    """
+    if not _is_configured():
+        return None
+
+    title = "Beli Nightly Refresh"
+    try:
+        category_id = await _get_first_category_id()
+
+        # Check whether an event with this title already exists.
+        async with httpx.AsyncClient() as client:
+            resp = await client.get(
+                f"{_base_url()}/api/app/get_schedule/v1",
+                params={"api_key": _api_key()},
+                timeout=10,
+            )
+            resp.raise_for_status()
+            existing = resp.json().get("rows", [])
+
+        if any(e.get("title") == title for e in existing):
+            logger.info("Cronicle: Beli nightly event already registered")
+            return next(e["id"] for e in existing if e.get("title") == title)
+
+        script = _make_beli_refresh_script()
+        async with httpx.AsyncClient() as client:
+            resp = await client.post(
+                f"{_base_url()}/api/app/create_event/v1",
+                json={
+                    "api_key": _api_key(),
+                    "title": title,
+                    "enabled": 1,
+                    "plugin": "shellplug",
+                    "target": "allgrp",
+                    "category": category_id,
+                    "timing": {"hours": [1], "minutes": [0]},
+                    "timezone": "UTC",
+                    "timeout": 60,
+                    "params": {"script": script},
+                    "notes": "Auto-created: triggers Beli nightly restaurant refresh",
+                },
+                timeout=10,
+            )
+            resp.raise_for_status()
+            data = resp.json()
+            if data.get("code") != 0:
+                raise RuntimeError(f"Cronicle create_event error: {data}")
+            event_id = data["id"]
+            logger.info("Cronicle: created Beli nightly event %s", event_id)
+            return event_id
+
+    except Exception:
+        logger.exception("Failed to register Beli nightly Cronicle event")
+        return None
+
+
+async def create_message_event(
+    msg_id: str, scheduled_at: datetime, recurrence: str = "none"
+) -> str | None:
+    """Create a Cronicle event for a scheduled message.
+
+    Uses one-shot timing for one-time messages, recurring timing otherwise.
+    Returns the Cronicle event ID, or None if Cronicle is not configured.
+    """
+    if not _is_configured():
+        return None
+
+    try:
+        category_id = await _get_first_category_id()
+        timing = _build_timing(scheduled_at, recurrence)
+        script = _make_trigger_script()
+
+        async with httpx.AsyncClient() as client:
+            resp = await client.post(
+                f"{_base_url()}/api/app/create_event/v1",
+                json={
+                    "api_key": _api_key(),
+                    "title": f"Scheduled Message {msg_id}",
+                    "enabled": 1,
+                    "plugin": "shellplug",
+                    "target": "allgrp",
+                    "category": category_id,
+                    "timing": timing,
+                    "timezone": "UTC",
+                    "timeout": 360,
+                    "params": {"script": script},
+                    "notes": f"Auto-created for scheduled message {msg_id} (recurrence={recurrence})",
+                },
+                timeout=10,
+            )
+            resp.raise_for_status()
+            data = resp.json()
+            if data.get("code") != 0:
+                raise RuntimeError(f"Cronicle create_event error: {data}")
+            event_id = data["id"]
+            logger.info(
+                "Cronicle: created %s event %s for message %s",
+                recurrence,
+                event_id,
+                msg_id,
+            )
+            return event_id
+
+    except Exception:
+        logger.exception(
+            "Failed to create Cronicle event for message %s — will not fire", msg_id
+        )
+        return None

utils/google_service.py

@@ -0,0 +1,177 @@
+"""Google Calendar and Gmail service using the Python SDK."""
+
+import asyncio
+import base64
+import os
+from typing import Any
+
+from google.auth.transport.requests import Request
+from google.oauth2.credentials import Credentials
+from googleapiclient.discovery import build
+
+SCOPES = [
+    "https://www.googleapis.com/auth/calendar.events",
+    "https://www.googleapis.com/auth/gmail.readonly",
+]
+
+
+class GoogleService:
+    def __init__(self):
+        token_path = os.getenv("GOOGLE_TOKEN_FILE", "/app/config/google-token.json")
+        if not os.path.exists(token_path):
+            raise FileNotFoundError(f"Google token file not found: {token_path}")
+
+        self._token_path = token_path
+        creds = Credentials.from_authorized_user_file(token_path, SCOPES)
+        if creds.expired and creds.refresh_token:
+            creds.refresh(Request())
+            with open(token_path, "w") as f:
+                f.write(creds.to_json())
+        self._creds = creds
+
+        self._calendar = build("calendar", "v3", credentials=creds)
+        self._gmail = build("gmail", "v1", credentials=creds)
+
+    async def get_calendar_events(
+        self,
+        time_min: str,
+        time_max: str,
+        calendar_id: str = "primary",
+        max_results: int = 20,
+    ) -> list[dict[str, Any]]:
+        def _fetch():
+            result = (
+                self._calendar.events()
+                .list(
+                    calendarId=calendar_id,
+                    timeMin=time_min,
+                    timeMax=time_max,
+                    maxResults=max_results,
+                    singleEvents=True,
+                    orderBy="startTime",
+                )
+                .execute()
+            )
+            return result.get("items", [])
+
+        return await asyncio.to_thread(_fetch)
+
+    async def create_calendar_event(
+        self,
+        summary: str,
+        start_time: str,
+        end_time: str,
+        description: str = "",
+        location: str = "",
+        calendar_id: str = "primary",
+    ) -> dict[str, Any]:
+        def _create():
+            body: dict[str, Any] = {
+                "summary": summary,
+                "start": {"dateTime": start_time},
+                "end": {"dateTime": end_time},
+            }
+            if description:
+                body["description"] = description
+            if location:
+                body["location"] = location
+            return (
+                self._calendar.events()
+                .insert(calendarId=calendar_id, body=body)
+                .execute()
+            )
+
+        return await asyncio.to_thread(_create)
+
+    async def search_emails(
+        self,
+        query: str,
+        max_results: int = 10,
+    ) -> list[dict[str, Any]]:
+        def _fetch():
+            result = (
+                self._gmail.users()
+                .messages()
+                .list(userId="me", q=query, maxResults=max_results)
+                .execute()
+            )
+            messages = result.get("messages", [])
+
+            emails = []
+            for msg in messages:
+                msg_data = (
+                    self._gmail.users()
+                    .messages()
+                    .get(
+                        userId="me",
+                        id=msg["id"],
+                        format="metadata",
+                        metadataHeaders=["From", "To", "Subject", "Date"],
+                    )
+                    .execute()
+                )
+                headers = {
+                    h["name"]: h["value"]
+                    for h in msg_data.get("payload", {}).get("headers", [])
+                }
+                emails.append(
+                    {
+                        "id": msg_data["id"],
+                        "from": headers.get("From", ""),
+                        "to": headers.get("To", ""),
+                        "subject": headers.get("Subject", ""),
+                        "date": headers.get("Date", ""),
+                        "snippet": msg_data.get("snippet", ""),
+                    }
+                )
+            return emails
+
+        return await asyncio.to_thread(_fetch)
+
+    async def get_email(self, message_id: str) -> dict[str, Any]:
+        def _fetch():
+            msg_data = (
+                self._gmail.users()
+                .messages()
+                .get(userId="me", id=message_id, format="full")
+                .execute()
+            )
+            headers = {
+                h["name"]: h["value"]
+                for h in msg_data.get("payload", {}).get("headers", [])
+            }
+
+            body = _extract_body(msg_data.get("payload", {}))
+
+            return {
+                "id": msg_data["id"],
+                "from": headers.get("From", ""),
+                "to": headers.get("To", ""),
+                "subject": headers.get("Subject", ""),
+                "date": headers.get("Date", ""),
+                "body": body[:4000],
+                "snippet": msg_data.get("snippet", ""),
+            }
+
+        return await asyncio.to_thread(_fetch)
+
+
+def _extract_body(payload: dict[str, Any]) -> str:
+    mime = payload.get("mimeType", "")
+    data = payload.get("body", {}).get("data", "")
+
+    if mime == "text/plain" and data:
+        return base64.urlsafe_b64decode(data).decode("utf-8", errors="replace")
+
+    if mime == "text/html" and data:
+        import html2text
+
+        html = base64.urlsafe_b64decode(data).decode("utf-8", errors="replace")
+        return html2text.html2text(html)
+
+    for part in payload.get("parts", []):
+        result = _extract_body(part)
+        if result:
+            return result
+
+    return ""

utils/homebox_mcp.py

@@ -0,0 +1,123 @@
+"""Homebox inventory tools, loaded from the homebox-mcp server over stdio.
+
+Admin-only: the tools are only compiled into the admin agent, and each call
+re-checks ``is_admin`` in the run config as a second line of defense.
+"""
+
+import logging
+import os
+
+from langchain_core.runnables import RunnableConfig
+from langchain_core.tools import BaseTool, StructuredTool
+
+logger = logging.getLogger(__name__)
+
+# Read + safe writes only. The server exposes ~39 tools; exposing all of them
+# would blow the local model's 32k context and allow deletes.
+HOMEBOX_TOOL_ALLOWLIST = (
+    "search_items",
+    "get_item",
+    "list_locations",
+    "location_contents",
+    "list_tags",
+    "warranties_expiring",
+    "inventory_stats",
+    "list_maintenance",
+    "create_item",
+    "move_item",
+    "log_maintenance",
+)
+
+DESCRIPTION_MAX_CHARS = 300
+ADMIN_ONLY_MESSAGE = "Homebox access is restricted to admin users."
+
+
+def _content_to_text(content) -> str:
+    """Flatten MCP content blocks into plain text for the LLM and persistence."""
+    if isinstance(content, str):
+        return content
+    if isinstance(content, (list, tuple)):
+        parts = []
+        for block in content:
+            if isinstance(block, dict):
+                parts.append(block.get("text") or f"[{block.get('type', 'content')}]")
+            else:
+                parts.append(str(block))
+        return "\n".join(parts)
+    return str(content)
+
+
+def _short_description(description: str) -> str:
+    description = " ".join((description or "").split())
+    if len(description) > DESCRIPTION_MAX_CHARS:
+        description = description[: DESCRIPTION_MAX_CHARS - 3].rstrip() + "..."
+    return description
+
+
+def wrap_admin_only(inner: BaseTool) -> StructuredTool:
+    """Rename an MCP tool to ``homebox_<name>`` and gate it on ``is_admin``."""
+
+    async def call(config: RunnableConfig, **kwargs) -> str:
+        if not (config or {}).get("configurable", {}).get("is_admin"):
+            return ADMIN_ONLY_MESSAGE
+        try:
+            result = await inner.coroutine(**kwargs)
+        except Exception as e:
+            return f"Homebox error: {e}"
+        content = result[0] if isinstance(result, tuple) else result
+        return _content_to_text(content)
+
+    return StructuredTool(
+        name=f"homebox_{inner.name}",
+        description=_short_description(inner.description),
+        args_schema=inner.args_schema,
+        coroutine=call,
+        metadata=inner.metadata,
+    )
+
+
+def select_homebox_tools(mcp_tools: list[BaseTool]) -> list[StructuredTool]:
+    """Keep allowlisted tools (in allowlist order) and wrap them admin-only."""
+    by_name = {t.name: t for t in mcp_tools}
+    missing = [name for name in HOMEBOX_TOOL_ALLOWLIST if name not in by_name]
+    if missing:
+        logger.warning(f"Homebox MCP: server is missing tools {missing}")
+    return [
+        wrap_admin_only(by_name[name])
+        for name in HOMEBOX_TOOL_ALLOWLIST
+        if name in by_name
+    ]
+
+
+async def load_homebox_tools() -> list[StructuredTool]:
+    """Start homebox-mcp and return the wrapped tools, or [] if unavailable."""
+    url = os.getenv("HOMEBOX_URL")
+    token = os.getenv("HOMEBOX_TOKEN")
+    if not url or not token:
+        logger.info("Homebox MCP not configured (HOMEBOX_URL/HOMEBOX_TOKEN unset)")
+        return []
+
+    try:
+        from langchain_mcp_adapters.client import MultiServerMCPClient
+
+        client = MultiServerMCPClient(
+            {
+                "homebox": {
+                    "transport": "stdio",
+                    "command": os.getenv("HOMEBOX_MCP_COMMAND", "homebox-mcp"),
+                    "args": [],
+                    "env": {
+                        "HOMEBOX_URL": url,
+                        "HOMEBOX_TOKEN": token,
+                        "PATH": os.environ.get("PATH", ""),
+                    },
+                }
+            }
+        )
+        tools = select_homebox_tools(await client.get_tools())
+    except Exception:
+        logger.exception("Homebox MCP failed to load; continuing without it")
+        return []
+
+    logger.info(f"Homebox MCP: loaded {len(tools)} tools")
+    return tools

utils/image_process.py

@@ -87,50 +87,20 @@ def _to_jpeg_bytes(file_bytes: bytes) -> bytes:
     return buf.getvalue()
 
 
-async def analyze_user_image(file_bytes: bytes) -> str:
-    """Analyze an image uploaded by a user and return a text description.
+def image_content_block(file_bytes: bytes) -> dict:
+    """Build an OpenAI-style image_url content part from raw image bytes.
 
-    Uses llama-server (OpenAI-compatible API) with vision support.
-    Falls back to OpenAI if llama-server is not configured.
+    Normalizes to JPEG so HEIC/HEIF and other formats work, then encodes as a
+    base64 data URL suitable for a multimodal chat message.
     """
     import base64
 
-    from openai import AsyncOpenAI
-
-    file_bytes = _to_jpeg_bytes(file_bytes)
-
-    llama_url = os.getenv("LLAMA_SERVER_URL")
-    if llama_url:
-        aclient = AsyncOpenAI(base_url=llama_url, api_key="not-needed")
-        model = os.getenv("LLAMA_MODEL_NAME", "llama-3.1-8b-instruct")
-    else:
-        aclient = AsyncOpenAI()
-        model = "gpt-4o-mini"
-
-    b64 = base64.b64encode(file_bytes).decode("utf-8")
-
-    response = await aclient.chat.completions.create(
-        model=model,
-        messages=[
-            {
-                "role": "system",
-                "content": "You are a helpful image analyst. Describe what you see in the image in detail. Be thorough but concise.",
-            },
-            {
-                "role": "user",
-                "content": [
-                    {"type": "text", "text": "Please describe this image in detail."},
-                    {
-                        "type": "image_url",
-                        "image_url": {
-                            "url": f"data:image/jpeg;base64,{b64}",
-                        },
-                    },
-                ],
-            },
-        ],
-    )
-    return response.choices[0].message.content
+    jpeg_bytes = _to_jpeg_bytes(file_bytes)
+    b64 = base64.b64encode(jpeg_bytes).decode("utf-8")
+    return {
+        "type": "image_url",
+        "image_url": {"url": f"data:image/jpeg;base64,{b64}"},
+    }
 
 
 if __name__ == "__main__":

utils/strip_markdown.py

@@ -1,5 +1,26 @@
 import re
 
+_THINK_RE = re.compile(r"<think>.*?</think>", re.DOTALL)
+
+
+def extract_agent_text(content) -> str:
+    """Extract plain text from a LangChain AIMessage content.
+
+    Handles both str and list-of-blocks content (the latter occurs with some
+    tool-using or vision models), and strips <think>...</think> tokens emitted
+    by reasoning models like Qwen3.
+    """
+    if isinstance(content, list):
+        text = "".join(
+            block.get("text", "")
+            if isinstance(block, dict) and block.get("type") != "thinking"
+            else (str(block) if not isinstance(block, dict) else "")
+            for block in content
+        )
+    else:
+        text = str(content) if content is not None else ""
+    return _THINK_RE.sub("", text).strip()
+
 
 def strip_markdown(text: str) -> str:
     """Strip markdown formatting from text for plain-text channels like iMessage."""

utils/ynab_service.py

@@ -235,6 +235,77 @@ class YNABService:
             else 0,
         }
 
+    def get_category_allocations(self, month: Optional[str] = None) -> dict[str, Any]:
+        """Get all category allocations (assigned and available) grouped by category group.
+
+        Args:
+            month: Month in YYYY-MM format (defaults to current month)
+
+        Returns:
+            Dictionary with all categories grouped, showing assigned and available amounts.
+        """
+        if not month:
+            month = datetime.now().strftime("%Y-%m-01")
+        else:
+            if len(month) == 7:
+                month = f"{month}-01"
+
+        month_date_obj = datetime.strptime(month, "%Y-%m-%d").date()
+
+        # Get category groups so we know which group each category belongs to
+        categories_response = self.categories_api.get_categories(self.budget_id)
+        category_group_map: dict[str, str] = {}
+        group_order: list[str] = []
+        for group in categories_response.data.category_groups or []:
+            if group.deleted or group.hidden:
+                continue
+            if group.name not in group_order:
+                group_order.append(group.name)
+            for cat in group.categories or []:
+                category_group_map[cat.id] = group.name
+
+        # Get month data for assigned/available amounts
+        month_response = self.months_api.get_plan_month(self.budget_id, month_date_obj)
+        month_data = month_response.data.month
+
+        groups: dict[str, list] = {g: [] for g in group_order}
+        total_assigned = 0.0
+        total_available = 0.0
+
+        for category in month_data.categories or []:
+            if category.deleted or category.hidden:
+                continue
+
+            assigned = category.budgeted / 1000
+            activity = category.activity / 1000
+            available = category.balance / 1000
+
+            group_name = category_group_map.get(category.id, "Other")
+            if group_name not in groups:
+                groups[group_name] = []
+
+            groups[group_name].append(
+                {
+                    "name": category.name,
+                    "assigned": round(assigned, 2),
+                    "activity": round(activity, 2),
+                    "available": round(available, 2),
+                }
+            )
+            total_assigned += assigned
+            total_available += available
+
+        return {
+            "month": month[:7],
+            "group_order": group_order,
+            "groups": groups,
+            "total_assigned": round(total_assigned, 2),
+            "total_available": round(total_available, 2),
+            "to_be_budgeted": round(month_data.to_be_budgeted / 1000, 2)
+            if month_data.to_be_budgeted
+            else 0,
+        }
+
     def get_spending_insights(self, months_back: int = 3) -> dict[str, Any]:
         """Generate insights about spending patterns and budget health.