diff --git a/src/flask/brew_advice.py b/src/flask/brew_advice.py index 222c21a..da914dc 100644 --- a/src/flask/brew_advice.py +++ b/src/flask/brew_advice.py @@ -67,6 +67,9 @@ temperature? - Should they rack, add more hops, raise the temperature, or wait? Rules: +- Advise about the reading that is marked as the one to advise on. It + may not be the most recent one - the brewer may be correcting an older + reading - and the advice must be about the reading they asked about. - If everything looks normal, say so briefly and say what to expect next. - If something looks wrong, say what it is and what to do about it. - Be specific to the numbers given. Do not give generic advice. @@ -81,8 +84,13 @@ Write two to five sentences of plain prose. """ -def _reading_lines(readings: list[dict]) -> str: - """The readings, as the compact list the model reads.""" +def _reading_lines(readings: list[dict], subject_id: str | None = None) -> str: + """The readings, as the compact list the model reads. + + The reading being advised about is marked, so the model does not have + to work out which one from the dates - the brewer may be correcting + an older reading. + """ lines = [] for r in readings: parts = [f"- {r.get('date')}: gravity {r.get('sg_decimal')} ({r.get('point_scale')})"] @@ -90,12 +98,19 @@ def _reading_lines(readings: list[dict]) -> str: parts.append(f"brewer's comment: {r['comment']}") if r.get("final"): parts.append("(marked as the FINAL reading)") + if subject_id is not None and str(r.get("_id")) == str(subject_id): + parts.append("<<< THIS IS THE READING TO ADVISE ON") lines.append(" ".join(parts)) return "\n".join(lines) -def build_prompt(run: dict, reading: dict, history: list[dict]) -> str: - """The user-message content for one observation.""" +def build_prompt(run: dict, reading: dict, history: list[dict], + corrected: bool = False) -> str: + """The user-message content for one observation. + + *corrected* says the reading has just been edited, so the advice + should take that into account. + """ name = run.get("canonical_name") or run.get("name") or "this brew" og = run.get("og_decimal") if og is None and run.get("og_points"): @@ -113,10 +128,23 @@ def build_prompt(run: dict, reading: dict, history: list[dict]) -> str: parts.append("") parts.append("Readings so far (oldest first):") - parts.append(_reading_lines(history)) + parts.append(_reading_lines(history, subject_id=reading.get("_id"))) parts.append("") - parts.append("Give advice for the most recent reading.") + # The reading being advised about is NAMED, not just described as + # "the most recent one". The brewer may be editing an older reading, + # and the advice is about THAT reading - so the prompt has to say + # which one, or the model advises about the newest instead. + parts.append( + f"Give advice about the reading dated {reading.get('date') or 'unknown'}" + f" (gravity {reading.get('sg_decimal')})," + + (" which the brewer has just corrected." if corrected else "") + ) + if corrected: + parts.append( + "It is the reading they have just corrected, so it may differ " + "from what it said before." + ) return "\n".join(parts) @@ -139,7 +167,7 @@ def call_ollama(prompt: str, *, url: str | None = None, model: str | None = None # allowance and the reply came back empty - which looked like a # broken model rather than a budget problem. 2000 leaves room for # the reasoning AND the advice. - "max_tokens": 2000, + "max_tokens": 3000, "stream": False, }).encode("utf-8") @@ -163,6 +191,7 @@ def call_ollama(prompt: str, *, url: str | None = None, model: str | None = None def generate_advice(run: dict, reading: dict, history: list[dict], *, + corrected: bool = False, url: str | None = None, model: str | None = None, timeout: float | None = None) -> tuple[bool, str]: """Generate the advice for one reading. @@ -171,7 +200,7 @@ def generate_advice(run: dict, reading: dict, history: list[dict], *, short note explaining what went wrong, so the reading carries a visible reason instead of silently having no advice. """ - prompt = build_prompt(run, reading, history) + prompt = build_prompt(run, reading, history, corrected=corrected) try: return True, call_ollama(prompt, url=url, model=model, timeout=timeout) except urllib.error.HTTPError as exc: @@ -197,14 +226,15 @@ def generate_advice(run: dict, reading: dict, history: list[dict], *, def generate_advice_in_background(run: dict, reading: dict, - history: list[dict], writer) -> None: + history: list[dict], writer, + corrected: bool = False) -> None: """Generate the advice and hand it to *writer* when it is ready. *writer* is called as ``writer(reading_id, ok, text)``. It runs in a background thread, so it must not touch the request context - the caller is responsible for giving it a way to reach Mongo. """ - ok, text = generate_advice(run, reading, history) + ok, text = generate_advice(run, reading, history, corrected=corrected) try: writer(reading.get("_id"), ok, text) except Exception: # noqa: BLE001 - never let the thread die noisily diff --git a/src/flask/mongo/repository.py b/src/flask/mongo/repository.py index 5aceb89..e0d31e8 100644 --- a/src/flask/mongo/repository.py +++ b/src/flask/mongo/repository.py @@ -19,6 +19,7 @@ from __future__ import annotations import logging import os +import threading from typing import Any from pymongo import MongoClient @@ -380,13 +381,30 @@ def _as_id(value: object) -> str: #: repository does too. _shared_repository: "BeerRepository | None" = None +#: Guards the check-then-set above. +#: +#: Without it, concurrent advice threads each saw ``None`` and each +#: built their own repository - so the fix reintroduced the very leak it +#: was meant to remove, just bounded by a race instead of by the number +#: of readings. Eight threads on a cold process built eight clients and +#: twenty-five threads. +_shared_repository_lock = threading.Lock() + def get_shared_repository() -> "BeerRepository": - """The process-wide repository, built on first use.""" + """The process-wide repository, built on first use. + + Thread-safe: the first caller builds it, the rest wait and get the + same one. + """ global _shared_repository if _shared_repository is None: - _shared_repository = BeerRepository() + with _shared_repository_lock: + # Checked again inside the lock - another thread may have + # built it while this one was waiting. + if _shared_repository is None: + _shared_repository = BeerRepository() return _shared_repository diff --git a/src/flask/views/brews.py b/src/flask/views/brews.py index bb85916..6b00c07 100644 --- a/src/flask/views/brews.py +++ b/src/flask/views/brews.py @@ -53,7 +53,7 @@ def advice_enabled() -> bool: return os.getenv("LLM_ADVICE", "1").strip().lower() not in {"0", "false", "no"} -def _start_advice(run_id: str, reading_id: str) -> None: +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 @@ -131,7 +131,7 @@ def _start_advice(run_id: str, reading_id: str) -> None: thread = threading.Thread( target=brew_advice.generate_advice_in_background, - args=(run_info, latest, readings, write), + args=(run_info, latest, readings, write, corrected), daemon=True, ) thread.start() @@ -208,6 +208,14 @@ def index(): "warning", ) 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(), @@ -221,7 +229,7 @@ def index(): # 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) + _start_advice(run_id, reading_id, corrected=was_existing) return redirect(url_for("brews.index")) runs = repo().runs() diff --git a/tests/test_brew_advice.py b/tests/test_brew_advice.py index 7098057..3a12d0f 100644 --- a/tests/test_brew_advice.py +++ b/tests/test_brew_advice.py @@ -56,6 +56,46 @@ def test_the_prompt_tells_the_model_not_to_call_a_reading_impossible(): assert "still be real" in brew_advice.SYSTEM_PROMPT +def test_the_prompt_names_the_reading_to_advise_about(): + """The reading being advised about must be NAMED. + + The brewer may be correcting an OLDER reading, so "the most recent + reading" is the wrong one - and the advice would be about the wrong + number. The reading is marked in the list and named in the + instruction. + """ + run = {"canonical_name": "Ale"} + older = {"_id": "old", "date": "2026-01-01", "sg_decimal": 1.042, + "point_scale": "1042 points", "comment": "", "final": False} + newer = {"_id": "new", "date": "2026-01-10", "sg_decimal": 1.030, + "point_scale": "1030 points", "comment": "", "final": False} + + prompt = brew_advice.build_prompt(run, older, [older, newer]) + + # The older one is the subject. + assert "THIS IS THE READING TO ADVISE ON" in prompt + # And it is named, with its date and gravity. + assert "2026-01-01" in prompt + assert "1.042" in prompt + # And NOT the newer one. + assert "1.03" in prompt # it is in the history, which is right + assert prompt.count("THIS IS THE READING TO ADVISE ON") == 1 + + +def test_the_prompt_says_when_a_reading_has_been_corrected(): + """A corrected reading may differ from what it said before.""" + run = {"canonical_name": "Ale"} + reading = {"_id": "r1", "date": "2026-01-01", "sg_decimal": 1.012, + "point_scale": "1012 points", "comment": "", "final": False} + + plain = brew_advice.build_prompt(run, reading, [reading]) + assert "just corrected" not in plain + + corrected = brew_advice.build_prompt(run, reading, [reading], corrected=True) + assert "just corrected" in corrected + assert "may differ" in corrected + + def test_the_prompt_marks_the_final_reading(): run = {"canonical_name": "Ale"} reading = {"date": "2026-01-20", "sg_decimal": 1.010, "point_scale": "1010 points", @@ -361,9 +401,10 @@ def test_the_advice_is_about_the_reading_that_was_saved(): captured = {} - def fake_generate(run, reading, history, writer=None): + def fake_generate(run, reading, history, writer=None, corrected=False): captured["reading"] = reading captured["history"] = history + captured["corrected"] = corrected return True, "THE ADVICE" original = brews_module.brew_advice.generate_advice_in_background @@ -396,6 +437,53 @@ def test_the_advice_is_about_the_reading_that_was_saved(): ) +def test_the_shared_repository_is_built_only_once(): + """Concurrent advice threads must not each build their own client. + + An unsynchronised check-then-set on a module global let eight threads + on a cold process each build their own repository - so the fix + reintroduced the very leak it was meant to remove, just bounded by a + race instead of by the number of readings. + """ + import threading + + import mongomock + import mongo.repository as R + + original_shared = R._shared_repository + built = [] + original_init = R.BeerRepository.__init__ + + def counting_init(self, *args, **kwargs): + built.append(1) + original_init(self, *args, **kwargs) + + R._shared_repository = None + R.BeerRepository.__init__ = counting_init + original_build = R._build_client + R._build_client = lambda: mongomock.MongoClient() + try: + instances = [] + barrier = threading.Barrier(8) + + def go(): + barrier.wait() # release all eight at once + instances.append(id(R.get_shared_repository())) + + threads = [threading.Thread(target=go) for _ in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + finally: + R._shared_repository = original_shared + R.BeerRepository.__init__ = original_init + R._build_client = original_build + + assert len(built) == 1, f"built {len(built)} repositories, expected 1" + assert len(set(instances)) == 1, "the threads got different repositories" + + def test_the_reading_id_cannot_be_used_to_inject_a_query(repo): """The reading id is coerced before it reaches Mongo.