feat(advice): LLM brewing advice on every observation (#14)

feat(advice): LLM brewing advice on every observation (#14)

src/flask/views/brews.py

@@ -2,15 +2,22 @@
 
 from __future__ import annotations
 
-from flask import (Blueprint, flash, jsonify, redirect, render_template,
-                   request, session, url_for)
+import logging
+import os
+import threading
+
+from flask import (Blueprint, current_app, flash, jsonify, redirect,
+                   render_template, request, session, url_for)
 from flask_wtf import FlaskForm
 from wtforms import (BooleanField, DateField, IntegerField, StringField,
                      SubmitField, TextAreaField)
 from wtforms.validators import DataRequired, Length, NumberRange, Optional
 
+import brew_advice
 from brewing import implausible_runs, is_plausible_sg, run_summary
 
+log = logging.getLogger(__name__)
+
 bp = Blueprint("brews", __name__)
 
 
@@ -34,6 +41,125 @@ def login_required(fn):
     return wrapper
 
 
+def advice_enabled() -> bool:
+    """Whether the LLM advice should be generated.
+
+    It is off in testing, so the suite never reaches out to a model, and
+    it can be turned off in production with ``LLM_ADVICE=0`` if the model
+    is misbehaving.
+    """
+    if current_app.config.get("TESTING"):
+        return False
+    return os.getenv("LLM_ADVICE", "1").strip().lower() not in {"0", "false", "no"}
+
+
+def _start_advice(run_id: str, reading_id: str, corrected: bool = False) -> None:
+    """Generate the LLM's advice for a reading, in the background.
+
+    The advice is a separate field on the reading, so the brewer's own
+    comment is never touched.
+
+    The background thread cannot use the request-scoped repository - that
+    client is closed when the request ends - so it uses the shared
+    process-wide one.
+    """
+    if not reading_id:
+        return
+    if not advice_enabled():
+        return
+
+    try:
+        repository = repo()
+        run = repository.run(run_id)
+        if run is None:
+            return
+        meta = repository.run_metadata(run_id)
+        summary = run_summary(run, og_override=meta.get("og_points"))
+        run_info = {
+            "canonical_name": meta.get("name") or run.display_name(),
+            "name": run.display_name(),
+            "og_points": summary.get("og_points"),
+            "og_decimal": summary.get("og_decimal"),
+            "tap_number": meta.get("tap_number"),
+            "notes": meta.get("notes", ""),
+        }
+
+        # The model's view of the run: every reading, with the gravity in
+        # both forms a brewer would recognise.
+        readings = []
+        for reading in run.readings:
+            points = reading.sg_points
+            readings.append({
+                "_id": str(reading._id),
+                "advice_token": reading.advice_token,
+                "date": reading.date.isoformat() if reading.date else "",
+                "sg": points,
+                "sg_decimal": reading.sg_decimal,
+                "point_scale": f"{points} points" if points is not None else "",
+                "comment": reading.comment or "",
+                "final": bool(reading.final_reading),
+            })
+
+        # The reading that is actually being advised about.  It is not
+        # necessarily the newest one: the brewer may be editing an older
+        # reading, and advice about the newest reading would be about the
+        # wrong number.
+        saved = next((r for r in readings if r.get("_id") == str(reading_id)), None)
+        if saved is None:
+            saved = dict(readings[-1]) if readings else {"date": "", "sg": None}
+        latest = dict(saved)
+        latest["_id"] = reading_id
+
+    except Exception:  # noqa: BLE001 - advice must never break the save
+        log.exception("could not gather the reading for the advice")
+        return
+
+    # The gravity and the editing token of the reading being advised
+    # about.  The write is conditional on both, so a slow advice thread
+    # cannot overwrite a correction with advice about the old value.
+    advice_sg = latest.get("sg")
+    advice_token = latest.get("advice_token")
+
+    def write(read_id: str, ok: bool, text: str) -> None:
+        # The client is SHARED, not built per reading.  A MongoClient
+        # spawns three monitor threads and pymongo does not reap them
+        # on close(), so building one per advice write accumulated three
+        # threads and a connection pool for every reading recorded.
+        #
+        # The shared client is created on first use and lives for the
+        # life of the process, which is what the request-scoped
+        # repository does too.
+        from mongo.repository import get_shared_repository
+
+        try:
+            shared = get_shared_repository()
+            written = shared.set_reading_advice(
+                read_id, text, ok,
+                expected_sg=advice_sg,
+                expected_token=advice_token,
+            )
+            if not written:
+                # The reading was corrected while the advice was being
+                # generated, so the advice no longer matches it.  That is
+                # the guard working - the fresh advice for the corrected
+                # reading will land in its own turn.
+                log.info(
+                    "the advice for reading %s was not stored: the reading "
+                    "has been edited since the advice was generated, so the "
+                    "advice no longer matches it",
+                    read_id,
+                )
+        except Exception:  # noqa: BLE001 - never let the thread die noisily
+            log.exception("could not store the generated advice")
+
+    thread = threading.Thread(
+        target=brew_advice.generate_advice_in_background,
+        args=(run_info, latest, readings, write, corrected),
+        daemon=True,
+    )
+    thread.start()
+
+
 def _last_reading_date(run) -> str | None:
     """The ISO date of the most recent reading in a run."""
     last = run.last_reading()
@@ -104,8 +230,17 @@ def index():
                     "(990-1200). It was saved anyway — double-check the reading.",
                     "warning",
                 )
-        repo().upsert_reading(
-            beer_run_id=str(form.beer_run_id.data),
+        run_id = str(form.beer_run_id.data)
+        # Whether the reading was already there: a re-submitted reading
+        # is a CORRECTION, and the advice should say so.
+        existing = repo().reading_docs_for_run(run_id)
+        was_existing = any(
+            str(doc.get("date")) == form.Date.data.isoformat()
+            for doc in existing
+        )
+
+        reading_id = repo().upsert_reading(
+            beer_run_id=run_id,
             beer_type=(form.beer_run_type.data or "").strip(),
             sg=sg_points,
             reading_date=form.Date.data.isoformat(),
@@ -113,6 +248,11 @@ def index():
             comment=(form.comment.data or "").strip(),
         )
         flash("Reading saved.", "success")
+
+        # The advice is generated in the background: recording a reading
+        # must not block on the model.  The page shows the reading straight
+        # away and the advice is there on a later visit.
+        _start_advice(run_id, reading_id, corrected=was_existing)
         return redirect(url_for("brews.index"))
 
     runs = repo().runs()