app.py

"""LifeOS — local-first personal assistant.

Gradio 6 Server mode: Gradio's API engine (queuing + SSE streaming) with a
fully hand-built frontend served from static/. All inference is local
(Nemotron-3-Nano-4B + nomic-embed through llama.cpp); the browser makes
zero external requests.
"""

import json
import logging
import os
import re
import tempfile
import threading
from collections.abc import Iterator

from fastapi import Request, UploadFile
from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from gradio import Server

import config
import engine
import memory as memory_store
import rag
from features import food, health, money, websearch

logger = logging.getLogger(__name__)

ROOT = os.path.dirname(os.path.abspath(__file__))
STATIC = os.path.join(ROOT, "static")

app = Server(title="LifeOS")


# ---------------------------------------------------------------- streaming
@app.api(name="food_recommend", concurrency_limit=1, stream_every=0.3)
def food_recommend(deals_json: str) -> Iterator[str]:
    deals = json.loads(deals_json) if deals_json else []
    mem = memory_store.load()
    user_input = food.build_food_input(deals, mem)
    yield from engine.run_domain("food", user_input)


@app.api(name="health_recommend", concurrency_limit=1, stream_every=0.3)
def health_recommend() -> Iterator[str]:
    mem = memory_store.load()
    yield from engine.run_domain("health", health.build_health_input(mem))


@app.api(name="money_review", concurrency_limit=1, stream_every=0.3)
def money_review(subs_json: str) -> Iterator[str]:
    subs = json.loads(subs_json) if subs_json else []
    mem = memory_store.load()
    if subs:
        memory_store.set_subscriptions(
            [{"name": s["name"], "cost": s["cost"], "last_used": s.get("last_charged", "")} for s in subs]
        )
        mem = memory_store.load()
    yield from engine.run_domain("money", money.build_money_input(subs, mem))


_REMEMBER_RE = re.compile(r"^\s*remember\b[,:]?\s*(.+)", re.IGNORECASE)


@app.api(name="chat", concurrency_limit=1, stream_every=0.3)
def chat(message: str, history_json: str, use_web: str = "", refs_json: str = "") -> Iterator[str]:
    history = json.loads(history_json) if history_json else []
    refs = json.loads(refs_json) if refs_json else []
    domains = [{"kitchen": "food"}.get(r, r) for r in refs] or None
    mem = memory_store.load()

    # explicit long-term memory writes: "remember ..."
    m = _REMEMBER_RE.match(message)
    if m:
        rag.remember(f"{mem['user_profile']['name']}: {m.group(1).strip()}", kind="fact")

    extra = ""
    if use_web == "1":
        extra = websearch.format_results(websearch.search_web(message))

    system, user = engine.build_prompt("chat", mem, message, domains=domains)
    messages = [system] + [h for h in history if h.get("role") in ("user", "assistant")][-8:] + [user]
    yield from engine.generate_stream(messages, domain="chat", extra_context=extra)


@app.api(name="goal_chat", concurrency_limit=1, stream_every=0.3)
def goal_chat(message: str, history_json: str, goal_json: str) -> Iterator[str]:
    """Socratic financial-goal coaching conversation."""
    history = json.loads(history_json) if history_json else []
    goal = json.loads(goal_json) if goal_json else {}
    mem = memory_store.load()

    user_input = message
    if goal:
        user_input = (
            f"GOAL UNDER DISCUSSION: {goal.get('title', '?')} — target ${goal.get('target_amount', '?')}, "
            f"saved ${goal.get('saved', 0)}, deadline {goal.get('deadline', '?')}\n\n{message}"
        )

    system, user = engine.build_prompt("goal", mem, user_input)
    messages = [system] + [h for h in history if h.get("role") in ("user", "assistant")][-8:] + [user]
    yield from engine.generate_stream(messages, domain="goal")


@app.api(name="local_deals", concurrency_limit=1, stream_every=0.3)
def local_deals(city: str) -> Iterator[str]:
    """Search the web for local grocery deals and stream a food recommendation."""
    mem = memory_store.load()
    city = city.strip() or mem["user_profile"].get("city", "Toronto")
    results = websearch.search_web(f"{city} grocery store weekly flyer deals")
    combined = "\n".join(f"{r['title']}\n{r['snippet']}" for r in results)
    deals = food.extract_deals_from_text(combined)
    if deals:
        user_input = (
            f"These grocery deals were found online for {city} this week:\n"
            + food.build_food_input(deals, mem)
        )
        yield from engine.run_domain("food", user_input)
    elif results:
        user_input = (
            f"No structured prices were parsed, but here is what a web search found "
            f"for grocery deals in {city}. Summarize where {mem['user_profile']['name']} "
            f"should shop this week and why:\n\n" + websearch.format_results(results)
        )
        yield from engine.run_domain("chat", user_input)
    else:
        yield (
            f"I couldn't reach the web to find deals in {city} right now. "
            "Try again later, or paste a flyer in the Food tab."
        )


@app.api(name="meal_analyze", concurrency_limit=1, stream_every=0.3)
def meal_analyze(ocr_text: str, meal_label: str) -> Iterator[str]:
    """Analyze OCR text from a food/receipt photo; optionally log the meal."""
    if meal_label.strip():
        memory_store.log_meal(meal_label.strip(), [])
    user_input = "OCR TEXT FROM PHOTO:\n" + ocr_text.strip()
    if meal_label.strip():
        user_input += f"\n\n(The user labeled this meal: {meal_label.strip()} — it has been logged.)"
    yield from engine.run_domain("meal_photo", user_input)


# ------------------------------------------------------------ non-streaming
@app.api(name="log_workout")
def log_workout(workout_type: str, duration: float) -> str:
    mem = memory_store.log_workout(workout_type, int(duration))
    rag.remember(
        f"{mem['user_profile']['name']} did a {workout_type} workout ({int(duration)} min) on {mem['workouts'][-1]['date']}",
        kind="event",
    )
    return json.dumps(mem)


@app.api(name="get_memory")
def get_memory() -> str:
    return json.dumps(memory_store.load())


@app.api(name="set_profile")
def set_profile(profile_json: str) -> str:
    fields = json.loads(profile_json) if profile_json else {}
    return json.dumps(memory_store.set_profile(fields))


@app.api(name="add_event")
def add_event(event_json: str) -> str:
    ev = json.loads(event_json) if event_json else {}
    return json.dumps(memory_store.add_event(ev))


@app.api(name="delete_event")
def delete_event(event_id: str) -> str:
    return json.dumps(memory_store.delete_event(event_id))


@app.api(name="set_schedule")
def set_schedule(days_json: str, time: str) -> str:
    days = json.loads(days_json) if days_json else []
    return json.dumps(memory_store.set_workout_schedule(days, time))


@app.api(name="set_payments")
def set_payments(payments_json: str) -> str:
    payments = json.loads(payments_json) if payments_json else []
    return json.dumps(memory_store.set_monthly_payments(payments))


@app.api(name="upsert_goal")
def upsert_goal(goal_json: str) -> str:
    goal = json.loads(goal_json) if goal_json else {}
    return json.dumps(memory_store.upsert_goal(goal))


@app.api(name="payment_impact", concurrency_limit=1, stream_every=0.3)
def payment_impact() -> Iterator[str]:
    """Stream a short note on how the current monthly payments affect the
    user's savings goal(s). Called after payments are saved."""
    mem = memory_store.load()
    if not mem.get("goals"):
        yield "Add a savings goal in the **Goals** tab to see how your monthly payments affect your timeline."
        return
    yield from engine.run_domain("payment_impact", "")


@app.api(name="delete_goal")
def delete_goal(goal_id: str) -> str:
    return json.dumps(memory_store.delete_goal(goal_id))


@app.api(name="edit_memory")
def edit_memory(section: str, items_json: str) -> str:
    """Replace a whitelisted memory section wholesale (manual editing)."""
    try:
        items = json.loads(items_json) if items_json else []
        return json.dumps(memory_store.set_section(section, items))
    except ValueError as e:
        return json.dumps({"error": str(e)})


@app.api(name="list_notes")
def list_notes() -> str:
    return json.dumps(rag.list_notes())


@app.api(name="update_note")
def update_note(note_id: str, text: str) -> str:
    rag.update_note(note_id, text)
    return json.dumps({"ok": True})


@app.api(name="delete_note")
def delete_note(note_id: str) -> str:
    rag.delete_note(note_id)
    return json.dumps({"ok": True})


@app.api(name="health_plan")
def health_plan() -> str:
    """Deterministic plan: free workout slots on preferred days over the next
    7 days that don't clash with calendar events (slot = time..time+1h)."""
    mem = memory_store.load()
    schedule = mem.get("workout_schedule", {"days": [], "time": "07:00"})
    events = memory_store.events_in_window(7, mem)
    return json.dumps({
        "schedule": schedule,
        "next_7_days_events": events,
        "recommendation_slots": health.plan_workout_slots(schedule, events),
    })


@app.api(name="reset_demo")
def reset_demo() -> str:
    """Load the sample persona. Only available when LIFEOS_DEMO=1 so it can't
    overwrite a real user's data in production."""
    if not config.DEMO:
        return json.dumps({"error": "Demo mode is off. Set LIFEOS_DEMO=1 to use sample data."})
    mem = memory_store.reset_to_seed()
    rag.reset_to_seed()
    return json.dumps(mem)


@app.api(name="reset_account")
def reset_account() -> str:
    """Wipe all stored data back to a blank slate (real-user reset)."""
    mem = memory_store.reset_to_empty()
    rag.reset_to_empty()
    return json.dumps(mem)


# ------------------------------------------------------------- file uploads
@app.post("/upload/flyer")
async def upload_flyer(file: UploadFile):
    suffix = os.path.splitext(file.filename or "flyer.pdf")[1] or ".pdf"
    with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
        tmp.write(await file.read())
        path = tmp.name
    try:
        deals = food.extract_deals(path)
    finally:
        os.unlink(path)
    return JSONResponse({"deals": deals})


@app.post("/upload/flyer_text")
async def upload_flyer_text(request: Request):
    body = await request.json()
    return JSONResponse({"deals": food.extract_deals_from_text(body.get("text", ""))})


@app.post("/upload/meal_photo")
async def upload_meal_photo(file: UploadFile):
    """Identify the food items in a meal/receipt photo using the local vision
    model. Falls back to OCR (for text-heavy receipts) if vision is
    unavailable. Returns {"text": ..., "source": "vision"|"ocr"} or {"error"}."""
    suffix = os.path.splitext(file.filename or "photo.jpg")[1] or ".jpg"
    with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
        tmp.write(await file.read())
        path = tmp.name
    try:
        try:
            text = engine.describe_food_image(path)
            return JSONResponse({"text": text, "source": "vision"})
        except Exception as e:  # vision model unavailable — try OCR for receipts
            logger.warning("Vision model failed for meal photo: %s", e)
        try:
            text = food.ocr_image(path)
            return JSONResponse({"text": text, "source": "ocr"})
        except Exception as e:  # neither backend available / unreadable image
            return JSONResponse(
                {"error": f"Couldn't read that photo (vision + OCR both failed): {e}"}
            )
    finally:
        os.unlink(path)


@app.post("/upload/transactions")
async def upload_transactions(file: UploadFile):
    text = (await file.read()).decode("utf-8", errors="replace")
    subs = money.detect_recurring(money.parse_transactions(text))
    return JSONResponse({"subscriptions": subs})


@app.get("/status")
async def status():
    """Model load state for the UI indicator: idle|loading|ready|error."""
    return JSONResponse({**engine.status(), "demo": config.DEMO})


# ----------------------------------------------------------------- frontend
class NoCacheStaticFiles(StaticFiles):
    """Serve assets with `Cache-Control: no-cache` so browsers always
    revalidate. Without this, StaticFiles sends no Cache-Control header and
    browsers apply heuristic freshness, serving a stale style.css/app.js for
    minutes/hours after an edit — which left the calendar modal rendered open
    on the landing page from an outdated stylesheet."""

    async def get_response(self, path, scope):
        response = await super().get_response(path, scope)
        response.headers["Cache-Control"] = "no-cache"
        return response


app.mount("/static", NoCacheStaticFiles(directory=STATIC), name="static")


@app.get("/")
async def index():
    return FileResponse(
        os.path.join(STATIC, "index.html"),
        headers={"Cache-Control": "no-cache"},
    )


if __name__ == "__main__":
    logging.basicConfig(
        level=os.environ.get("LIFEOS_LOG_LEVEL", "INFO"),
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    )
    # Warm the models in the background so the UI is reachable immediately
    # while the GGUFs download/load on first boot.
    threading.Thread(target=engine.warmup, daemon=True).start()
    # config.HOST is 0.0.0.0 on Spaces, else None so Gradio binds loopback.
    app.launch(server_name=config.HOST, server_port=config.PORT)