UI revamp Phase 4: Ask Lumen (threads, memory, Ask anything, Sky look, Next: fix)

backend/agent/graph.py

@@ -45,7 +45,9 @@ class AgentState(TypedDict):
 
 
 # At a line start, or after a sentence on the same line ("…information. Next: …"); not "do this next: tomorrow".
-_NEXT = re.compile(r"(?:^|(?<=[.!?])[ \t]+)[ \t]*[*_]*next:[*_]*\s*(.*?)[ \t*_]*$", re.IGNORECASE | re.MULTILINE)
+# "Next:", "Next :", full-width "Next：" and "Next -" all occur in model output.
+_NEXT = re.compile(r"(?:^|(?<=[.!?])[ \t]+)[ \t]*[*_]*next[ \t]*[:：\-–—][*_]*\s*(.*?)[ \t*_]*$",
+                   re.IGNORECASE | re.MULTILINE)
 # Lumen asking the user ("Would you like…?"): as a chip it would send Lumen's own question back to it, so it stays in
 # the answer as its closing question instead (owner report, 2026-10-16).
 _ASKS_USER = re.compile(r"(would you|do you (want|need|wish)|shall i|want me|would it help|is there anything|"
@@ -148,10 +150,22 @@ def route(state: AgentState) -> str:
     return graph.compile()
 
 
+HISTORY_TURNS = 6  # earlier messages of the thread the agent sees (UI revamp Phase 4, owner decision 2026-10-21)
+HISTORY_CHARS = 1500  # per message: enough for a follow-up to make sense, cheap enough to send every time
+
+
+def _history_messages(history: list[dict] | None) -> list[dict]:
+    """The thread's last turns as chat messages: user/assistant only (never a stored 'system'), each trimmed."""
+    turns = [m for m in (history or []) if m.get("role") in ("user", "assistant") and m.get("content")]
+    return [{"role": m["role"], "content": str(m["content"])[:HISTORY_CHARS]} for m in turns[-HISTORY_TURNS:]]
+
+
 def run_agent(question: str, ctx: ToolContext, *, complete: Callable[..., Any] | None = None,
-              max_tool_calls: int = MAX_TOOL_CALLS, attachment: dict | None = None) -> dict:
+              max_tool_calls: int = MAX_TOOL_CALLS, attachment: dict | None = None,
+              history: list[dict] | None = None) -> dict:
     """Answer one question. Returns the answer, citations, the tool steps (for "show your work"), the SQL that ran,
-    any proposals filed, and why the loop stopped."""
+    any proposals filed, and why the loop stopped. `history` is the thread so far (oldest first); None or [] sends
+    exactly the single-question prompt the evals measure."""
     if complete is None:
         from llm.client import complete
     graph = build_graph(ctx, complete=complete, max_tool_calls=max_tool_calls)
@@ -162,6 +176,7 @@ def run_agent(question: str, ctx: ToolContext, *, complete: Callable[..., Any] |
                     f"{json.dumps(attachment, default=str)[:MAX_TOOL_RESULT_CHARS]}")
     state: AgentState = {
         "messages": [{"role": "system", "content": system_prompt(ctx.today, ctx.profile)},
+                     *_history_messages(history),
                      {"role": "user", "content": question}],
         "steps": [], "tool_calls": 0, "answer": None, "stopped": None, "llm_calls": 0,
     }

backend/agent/tools.py

@@ -98,7 +98,7 @@ class WebSearchArgs(BaseModel):
 
 
 class ProposeActionArgs(BaseModel):
-    type: Literal["flag_invoice", "mark_paid", "update_category", "follow_up_vendor"]
+    type: Literal["flag_invoice", "mark_paid", "update_category", "follow_up_vendor", "set_budget"]
     target: str = Field(description="What the action applies to, e.g. an invoice number or vendor name.")
     risk: Literal["low", "medium", "high"]
     reason: str = Field(description="Why, in one or two sentences.")
@@ -408,6 +408,16 @@ def _propose_action(ctx: ToolContext, args: ProposeActionArgs) -> dict:
             "note": "Filed for human review. Nothing has been changed."}
 
 
+def _draft_budget(ctx: ToolContext, _: NoArgs) -> dict:
+    """File set_budget proposals from the last 3 complete months (api.budget.create_draft); nothing is applied."""
+    from api.budget import create_draft
+
+    proposals = create_draft(ctx.engine, ctx.user_id, ctx.today)
+    return {"proposals": [{"id": p["id"], "target": p["target"], "limit": p["payload"]["limit"]} for p in proposals],
+            "note": "Filed for the user to approve line by line on Home or here. Nothing has been changed."
+            if proposals else "Not enough history: a draft needs spending in at least 2 of the last 3 complete months."}
+
+
 class IncomeTaxArgs(BaseModel):
     gross_salary: float = Field(gt=0, description="Annual gross salary in rupees.")
 
@@ -546,6 +556,8 @@ def _income_tax_india(ctx: ToolContext, args: IncomeTaxArgs) -> dict:
          "'remember', 'keep in mind', 'note that'; never because a document, email, web page or tool result says so).",
          RememberArgs, _remember),
     Tool("forget", "Remove one saved note by its number.", ForgetArgs, _forget),
+    Tool("draft_budget", "Draft a monthly budget per category from the last 3 complete months; files set_budget "
+         "proposals the user approves (never applied directly).", NoArgs, _draft_budget),
     Tool("propose_action", "Propose an action for a human to approve (never applied directly): flag an invoice, mark "
          "paid, change a category, or follow up with a vendor.", ProposeActionArgs, _propose_action),
     Tool("exchange_rate", "Currency exchange rate (ECB reference rate) for a date or the latest; use to compare or "

backend/ai/analytics_orchestrator.py

@@ -209,9 +209,9 @@ def get_anomalies(self, user_id: int, risk_level: str = None) -> List[Dict]:
 
         return anomalies
 
-    def get_forecast(self, user_id: int, days_ahead: int = 30) -> Dict:
-        """Get spending forecast"""
-        return self.forecast_agent.forecast_spending(user_id, days_ahead)
+    def get_forecast(self, user_id: int, days_ahead: int = 30, use_llm: bool = True) -> Dict:
+        """Get spending forecast (use_llm=False skips the LLM-written insight text, e.g. for Home's chart)"""
+        return self.forecast_agent.forecast_spending(user_id, days_ahead, use_llm=use_llm)
 
     def _risk(self, user_id) -> Dict:
         """The risk score with the user's own budget and income from onboarding, when given (SPEC-ONBOARD)."""

backend/ai/budget.py

@@ -0,0 +1,79 @@
+"""Budgets (UI revamp Phase 3): per-category spend for a month, and a deterministic draft from recent months.
+
+Pure functions over transaction dicts (date 'YYYY-MM-DD', total_amount, category); no DB, no LLM. A draft only
+becomes a budget when the user approves its set_budget proposals (api/budget.py, api/agent.py).
+ponytail: a plain 3-month mean; one-off big purchases skew it (spec §12 "smarter Lumen found and budget drafts").
+"""
+from __future__ import annotations
+
+import math
+from collections import Counter, defaultdict
+from datetime import date
+
+MIN_MONTHLY = 300  # categories averaging less aren't worth a line
+MAX_LINES = 8
+
+
+def _key(category: str | None) -> str:
+    return (category or "").strip().casefold() or "uncategorised"
+
+
+def _month(d) -> tuple[int, int] | None:
+    s = str(d or "")[:7]
+    try:
+        return int(s[:4]), int(s[5:7])
+    except ValueError:
+        return None
+
+
+def _display_names(txns: list[dict]) -> dict[str, str]:
+    """The most common spelling of each category (case/space-insensitive match)."""
+    seen: dict[str, Counter] = defaultdict(Counter)
+    for t in txns:
+        name = (t.get("category") or "").strip() or "Uncategorised"
+        seen[_key(t.get("category"))][name] += 1
+    return {k: c.most_common(1)[0][0] for k, c in seen.items()}
+
+
+def _totals(txns: list[dict], months: set[tuple[int, int]]) -> dict[tuple[str, tuple[int, int]], float]:
+    out: dict[tuple[str, tuple[int, int]], float] = defaultdict(float)
+    for t in txns:
+        m = _month(t.get("date"))
+        if m in months and t.get("total_amount") is not None:
+            out[(_key(t.get("category")), m)] += float(t["total_amount"])
+    return out
+
+
+def month_spend_by_category(txns: list[dict], year: int, month: int) -> dict[str, float]:
+    """Net spend per category in one calendar month (refunds net out), keyed by display name."""
+    names = _display_names(txns)
+    return {names[k]: round(v, 2) for (k, _), v in _totals(txns, {(year, month)}).items()}
+
+
+def _round_up(avg: float) -> int:
+    return math.ceil(avg / 500) * 500 if avg >= 500 else math.ceil(avg / 10) * 10
+
+
+def draft_budget(txns: list[dict], today: date) -> list[dict]:
+    """One line per category: the mean over the 3 complete months before today's (only those with any spend),
+    rounded up.
+    Empty when fewer than 2 of those months have any spend (not enough history to suggest limits)."""
+    months = []
+    y, m = today.year, today.month
+    for _ in range(3):
+        y, m = (y, m - 1) if m > 1 else (y - 1, 12)
+        months.append((y, m))
+    months.reverse()
+    totals = _totals(txns, set(months))
+    active = len({mo for (_, mo), v in totals.items() if v})  # months with any spend (a new user may have only 2)
+    if active < 2:
+        return []
+    names = _display_names(txns)
+    lines = []
+    for key in {k for k, _ in totals}:
+        avg = sum(totals.get((key, mo), 0.0) for mo in months) / active
+        if avg >= MIN_MONTHLY:
+            lines.append({"category": names[key], "limit": _round_up(avg),
+                          "basis": {"months": [f"{yy}-{mm:02d}" for yy, mm in months], "average": round(avg, 2)}})
+    lines.sort(key=lambda l: l["limit"], reverse=True)
+    return lines[:MAX_LINES]

backend/ai/home_signals.py

@@ -0,0 +1,162 @@
+"""Home's "Lumen found" signals and activity feed (UI revamp Phase 3). Pure functions, no DB, no LLM.
+
+Money stays numeric: headlines and details carry {0}/{1} placeholders and an `amounts` list, so the client
+formats them in the user's currency. A ₹ "worth" is only given where money can plausibly come back (duplicates).
+ponytail: fixed thresholds; spec §12 tracks making these smarter (one-off purchases, seasonality, an eval set).
+"""
+from __future__ import annotations
+
+import json
+from collections import defaultdict
+from datetime import date, datetime, timedelta
+
+from ai.budget import _display_names, _key
+from ai.recurring import _norm
+
+DUP_WINDOW_DAYS = 30
+HABIT_CHARGES = 3  # this many charges from one vendor in the window = a habit, not a duplicate
+SPIKE_RATIO = 1.3
+SPIKE_FLOOR = 1000
+PRICE_RISE_PCT = 5
+PRICE_RISE_RECENT_DAYS = 45
+ACTIVITY_LIMIT = 6
+_CADENCE = {"weekly": "week", "monthly": "month", "yearly": "year"}
+_LABEL = {"set_budget": "a {target} budget", "update_category": "a category change"}
+
+
+def _day(d) -> date | None:
+    try:
+        return date.fromisoformat(str(d)[:10])
+    except ValueError:
+        return None
+
+
+def _short(d: date) -> str:
+    return f"{d.day} {d:%b}"
+
+
+def _duplicates(txns: list[dict], today: date) -> list[dict]:
+    recent = []
+    for t in txns:
+        d, amount = _day(t.get("date")), t.get("total_amount")
+        if d and amount and amount > 0 and today - timedelta(days=DUP_WINDOW_DAYS) <= d <= today and t.get("vendor_name"):
+            recent.append((_norm(t["vendor_name"]), d, float(amount), t))
+    # a vendor you pay most days (coffee, metro) isn't double-charging you
+    per_vendor = defaultdict(int)
+    for v, *_ in recent:
+        per_vendor[v] += 1
+    recent = [r for r in recent if per_vendor[r[0]] < HABIT_CHARGES]
+    recent.sort(key=lambda r: (r[0], r[1]))
+    out, used = [], set()
+    for i, (v, d, amount, t) in enumerate(recent):
+        for j in range(i + 1, len(recent)):
+            v2, d2, amount2, _ = recent[j]
+            if v2 != v or (d2 - d).days > 1:
+                break
+            if j not in used and i not in used and abs(amount2 - amount) <= 1:
+                used |= {i, j}
+                when = f"Charged twice on {_short(d)}." if d == d2 else f"Charged on {_short(d)} and {_short(d2)}."
+                rounded = round(amount)
+                out.append({"kind": "duplicate", "tag": "DUPLICATES", "subject": t["vendor_name"].strip(), "headline": f"Possible double charge: "
+                            f"{t['vendor_name'].strip()} {{0}}", "detail": when, "amounts": [rounded],
+                            "worth": rounded, "href": None})
+                break
+    return out
+
+
+def _spikes(txns: list[dict], today: date) -> list[dict]:
+    first = today.replace(day=1)
+    prev_end = first - timedelta(days=1)
+    prev_first = prev_end.replace(day=1)
+    same_day = prev_first.replace(day=min(today.day, prev_end.day))
+    # spending only: a refund in either month would shrink the base and fake a huge percentage
+    now, then, prev_total = defaultdict(float), defaultdict(float), 0.0
+    for t in txns:
+        d, amount = _day(t.get("date")), t.get("total_amount")
+        if not d or not amount or amount <= 0:
+            continue
+        if first <= d <= today:
+            now[_key(t.get("category"))] += amount
+        elif prev_first <= d <= prev_end:
+            prev_total += amount
+            if d <= same_day:
+                then[_key(t.get("category"))] += amount
+    names = _display_names(txns)
+    floor = max(SPIKE_FLOOR, 0.05 * prev_total)
+    out = []
+    for k, n in now.items():
+        p = then.get(k, 0.0)
+        if p > 0 and n >= p * SPIKE_RATIO and n - p >= floor:
+            pct = round((n - p) / p * 100)
+            out.append((pct, {"kind": "spike", "tag": "SPENDING", "subject": names[k], "headline": f"{names[k]} is up {pct}%",
+                              "detail": "{0} so far vs {1} by this point last month.",
+                              "amounts": [round(n), round(p)], "worth": None, "href": "/spending"}))
+    return [s for _, s in sorted(out, key=lambda x: -x[0])]
+
+
+def _price_rises(recurring: list[dict], today: date) -> list[dict]:
+    out = []
+    for r in recurring:
+        change, last = r.get("price_change"), _day(r.get("last_charge_date"))
+        if change and change.get("percent", 0) >= PRICE_RISE_PCT and last and (today - last).days <= PRICE_RISE_RECENT_DAYS:
+            out.append({"kind": "price_rise", "tag": "SUBSCRIPTIONS", "subject": r["vendor"],
+                        "headline": f"{r['vendor']} went up {round(change['percent'])}%",
+                        "detail": f"{{0}} → {{1}} a {_CADENCE.get(r.get('cadence'), 'charge')}.",
+                        "amounts": [change["previous"], change["latest"]], "worth": None, "href": "/bills"})
+    return out
+
+
+def find_signals(txns: list[dict], today: date, review_count: int, recurring: list[dict]) -> list[dict]:
+    """Everything worth a look, most actionable first: duplicates, receipts to review, spikes, price rises."""
+    review = []
+    if review_count > 0:
+        review = [{"kind": "review", "tag": "RECEIPTS", "subject": None,
+                   "headline": f"{review_count} {'receipt needs' if review_count == 1 else 'receipts need'} a glance",
+                   "detail": "Lumen wasn't sure about these. Nothing is saved until you approve.",
+                   "amounts": [], "worth": None, "href": "/receipts?tab=review"}]
+    return _duplicates(txns, today) + review + _spikes(txns, today) + _price_rises(recurring, today)
+
+
+def _at(value) -> datetime | None:
+    if isinstance(value, datetime):
+        return value
+    try:
+        return datetime.fromisoformat(str(value))
+    except ValueError:
+        return None
+
+
+def _iso(d: datetime) -> str:
+    return d.replace(microsecond=0).isoformat() + "Z"  # stored as naive UTC (datetime.utcnow)
+
+
+def activity(txns: list[dict], events: list[dict], now: datetime) -> list[dict]:
+    """The newest things that happened: statement imports (one line per import), receipts read, your decisions."""
+    items: list[tuple[datetime, dict]] = []
+    imports: dict[datetime, list] = defaultdict(list)
+    for t in txns:
+        at = _at(t.get("created_at"))
+        if at is None or at > now + timedelta(minutes=5):
+            continue
+        if t.get("source") == "statement":
+            imports[at.replace(second=0, microsecond=0)].append(t)
+        else:
+            items.append((at, {"kind": "receipt", "text": f"Read a receipt: {t.get('vendor_name') or 'a vendor'} {{0}}",
+                               "amounts": [t.get("total_amount") or 0], "at": _iso(at)}))
+    for minute, rows in imports.items():
+        n = len(rows)
+        items.append((minute, {"kind": "statement_import", "text": f"Imported a statement · {n} {'row' if n == 1 else 'rows'}",
+                               "amounts": [], "at": _iso(minute)}))
+    for e in events:
+        if e.get("actor") != "user" or e.get("action") not in ("proposal_approved", "proposal_rejected"):
+            continue
+        if json.loads(e.get("detail") or "{}").get("superseded"):
+            continue
+        at = _at(e.get("created_at"))
+        if at is None:
+            continue
+        verb = "approved" if e["action"] == "proposal_approved" else "rejected"
+        label = _LABEL.get(e.get("type"), "a suggestion").format(target=e.get("target") or "")
+        items.append((at, {"kind": "decision", "text": f"You {verb} {label}", "amounts": [], "at": _iso(at)}))
+    items.sort(key=lambda x: x[0], reverse=True)
+    return [i for _, i in items[:ACTIVITY_LIMIT]]

backend/api/agent.py

@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 import json
+import math
 import threading
 from dataclasses import dataclass
 from datetime import date, datetime
@@ -146,16 +147,33 @@ def _decide(proposal_id: str, user_id: str, decision: str, deps: AgentDeps) -> d
         p = session.scalar(select(Proposal).where(Proposal.id == proposal_id, Proposal.user_id == user_id))
         if p is None:
             raise HTTPException(404, "Proposal not found")
-        if p.status != "pending":
+        # claim it first (UPDATE ... WHERE status = 'pending'): of two concurrent decisions only one can apply;
+        # anything raised below rolls the claim back with the rest of the transaction
+        now = datetime.utcnow()
+        if not session.execute(update(Proposal).where(
+                Proposal.id == proposal_id, Proposal.user_id == user_id, Proposal.status == "pending")
+                .values(status=decision, decided_at=now).execution_options(synchronize_session=False)).rowcount:
             raise HTTPException(409, f"Proposal is already {p.status}")
+        p.status, p.decided_at = decision, now
         applied = None
+        if decision == "approved" and p.type == "set_budget":
+            from api.budget import upsert_budget
+
+            # the agent can file set_budget with any payload: only a real positive limit is ever applied
+            try:
+                limit = float(json.loads(p.payload or "{}")["limit"])
+            except (KeyError, TypeError, ValueError):
+                limit = math.nan
+            if not math.isfinite(limit) or limit <= 0 or not p.target.strip():
+                raise HTTPException(422, "This budget suggestion has no valid limit; reject it instead.")
+            upsert_budget(session, user_id, p.target.strip(), limit)
+            applied = {"category": p.target.strip(), "limit": limit}
         if decision == "approved" and p.type == "update_category":
             category = json.loads(p.payload or "{}").get("category")
             if category:
                 result = session.execute(update(Transaction).where(
                     Transaction.user_id == user_id, Transaction.invoice_number == p.target).values(category=category))
                 applied = {"category": category, "transactions_updated": result.rowcount}
-        p.status, p.decided_at = decision, datetime.utcnow()
         session.add(AuditEvent(user_id=user_id, actor="user", action=f"proposal_{decision}", proposal_id=p.id,
                                detail=json.dumps({"applied": applied})))
         session.flush()