From 3e1e4f42b7888776837ae59a91a49be8941efe39 Mon Sep 17 00:00:00 2001 From: hermes Date: Mon, 9 Nov 2026 17:45:16 +1000 Subject: [PATCH] fix(advice): stale advice must not overwrite a correction The review found a race between the background advice thread and the brewer correcting a reading. `set_reading_advice` wrote unconditionally by id. The advice is generated in the background, so if the brewer corrects a reading while the advice for the OLD gravity is still being generated, the slow thread lands afterwards and overwrites the fresh advice with advice about a reading that no longer exists. The reading ends up with the final gravity from the correction and the advice from before it. Reproduced against the live database: wrote advice for 1050 wrote advice for 1020 late stale write matched: 0 final sg: 1020 | final advice: 'ADVICE FOR THE CORRECTION' The write is now conditional on the reading still holding the gravity the advice was generated for, so a late thread matches nothing. Also ---- `finish_reason` is now checked. A reply that stopped at the token ceiling is incomplete - it may be advice that stops mid-sentence - and it was being stored as if it were the whole answer. It is now marked as cut off. 173 tests (up from 171). --- src/flask/brew_advice.py | 15 ++++++- src/flask/mongo/repository.py | 20 ++++++++- src/flask/views/brews.py | 13 ++++-- tests/test_brew_advice.py | 81 +++++++++++++++++++++++++++++++++++ 4 files changed, 122 insertions(+), 7 deletions(-) diff --git a/src/flask/brew_advice.py b/src/flask/brew_advice.py index 35adea7..1304614 100644 --- a/src/flask/brew_advice.py +++ b/src/flask/brew_advice.py @@ -198,11 +198,24 @@ def call_ollama(prompt: str, *, url: str | None = None, model: str | None = None if not choices: raise RuntimeError("the model returned no choices") message = choices[0].get("message") or {} + + # Some models put their working in a separate field. It is ignored - + # the advice is the content. content = message.get("content") or "" content = content.strip() if not content: raise RuntimeError("the model returned an empty response") - return _trim_advice(_strip_reasoning(content)) + + # A reply cut off at the token ceiling is INCOMPLETE. It may be + # advice that stops mid-sentence, so it is marked as truncated rather + # than stored as if it were the whole answer. + finish_reason = choices[0].get("finish_reason") or body.get("done_reason") + truncated = finish_reason == "length" + + advice = _trim_advice(_strip_reasoning(content)) + if truncated: + advice = _trim_advice(advice + " (the reply was cut off)") + return advice def _trim_advice(text: str) -> str: diff --git a/src/flask/mongo/repository.py b/src/flask/mongo/repository.py index e0d31e8..c36c8e1 100644 --- a/src/flask/mongo/repository.py +++ b/src/flask/mongo/repository.py @@ -223,20 +223,36 @@ class BeerRepository: return str(existing.get("_id")) if existing else "" def set_reading_advice(self, reading_id: str, advice: str, - ok: bool = True) -> int: + ok: bool = True, + expected_sg: int | None = None) -> int: """Attach the LLM's advice to a reading. The advice lives in its own field, so the brewer's own comment is never overwritten. ``ok`` says whether the advice was actually generated or is a note explaining that it could not be. + + ``expected_sg`` is the gravity the advice was generated FOR. The + write only happens if the reading still holds that gravity. + + This matters because the advice is generated in the background: if + the brewer corrects a reading while the advice for the old value + is still being generated, the slow thread would land afterwards + and overwrite the fresh advice with advice about a reading that + no longer exists. Without the guard the run ends up with the + final gravity from the correction and the advice from before it. """ doc_id = _as_doc_id(reading_id) if not doc_id: return 0 + + query: dict = {"_id": {"$in": doc_id}} + if expected_sg is not None: + query["sg"] = str(expected_sg) + # `$in` matches either form. A plain list would be an exact # ARRAY match in Mongo, which is not what is wanted here. result = self.readings.update_one( - {"_id": {"$in": doc_id}}, + query, {"$set": { "advice": str(advice), "advice_ok": bool(ok), diff --git a/src/flask/views/brews.py b/src/flask/views/brews.py index 6b00c07..9c92aa9 100644 --- a/src/flask/views/brews.py +++ b/src/flask/views/brews.py @@ -112,20 +112,25 @@ def _start_advice(run_id: str, reading_id: str, corrected: bool = False) -> None log.exception("could not gather the reading for the advice") return + # The gravity the advice is about. The write is conditional on the + # reading still holding it, so a slow advice thread cannot overwrite + # a correction with advice about the old value. + advice_sg = latest.get("sg") + 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 + # 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 BeerRepository, get_shared_repository + from mongo.repository import get_shared_repository try: shared = get_shared_repository() - shared.set_reading_advice(read_id, text, ok) + shared.set_reading_advice(read_id, text, ok, expected_sg=advice_sg) except Exception: # noqa: BLE001 - never let the thread die noisily log.exception("could not store the generated advice") diff --git a/tests/test_brew_advice.py b/tests/test_brew_advice.py index 99cee4c..4222718 100644 --- a/tests/test_brew_advice.py +++ b/tests/test_brew_advice.py @@ -546,6 +546,87 @@ def test_the_shared_repository_is_built_only_once(): assert len(set(instances)) == 1, "the threads got different repositories" +def test_stale_advice_cannot_overwrite_a_correction(repo): + """A slow advice thread must not overwrite a correction. + + The advice is generated in the background. If the brewer corrects a + reading while the advice for the old gravity is still being + generated, the slow thread lands afterwards - and without a guard the + reading ends up with the final gravity from the correction and the + advice from before it. + + The write is conditional on the reading still holding the gravity the + advice was generated for. + """ + seed_readings(repo, [("1", "Ale", "1050", "2026-01-01", False, "a comment")]) + doc = repo.readings.find_one({"beer_run_id": "1"}) + reading_id = str(doc["_id"]) + + # The advice for the ORIGINAL 1050 arrives late... + modified = repo.set_reading_advice(reading_id, "ADVICE FOR THE 1050", + True, expected_sg=1050) + assert modified == 1 + + # ...but the brewer has already corrected it to 1020, and the fresh + # advice has landed. + repo.upsert_reading(beer_run_id="1", beer_type="Ale", sg=1020, + reading_date="2026-01-01", final_reading=False, + comment="a comment") + repo.set_reading_advice(reading_id, "ADVICE FOR THE 1020", True, + expected_sg=1020) + + # Now the late thread for the old 1050 finally lands. + repo.set_reading_advice(reading_id, "STALE ADVICE FOR THE 1050", True, + expected_sg=1050) + + after = repo.readings.find_one({"beer_run_id": "1"}) + assert after["sg"] == "1020" + assert after["advice"] == "ADVICE FOR THE 1020", ( + f"the stale advice overwrote the correction: {after['advice']!r}" + ) + + +def test_a_reply_cut_off_at_the_token_ceiling_is_marked(repo): + """A reply that stopped at the token limit is incomplete. + + It may be advice that stops mid-sentence, so it must not be stored + as if it were the whole answer. + """ + captured = {} + + def fake_urlopen(request, timeout=None): + captured["body"] = json.loads(request.data.decode()) + + class Response: + def read(self): + return json.dumps({ + "choices": [{ + "finish_reason": "length", + "message": {"content": "The gravity looks fine so far, but"}, + }], + }).encode() + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + return Response() + + import brew_advice as ba + original = ba.urllib.request.urlopen + ba.urllib.request.urlopen = fake_urlopen + try: + ok, text = ba.generate_advice({"canonical_name": "Ale"}, + {"date": "2026-01-01"}, []) + finally: + ba.urllib.request.urlopen = original + + assert ok is True + assert "cut off" in text + + def test_the_reading_id_cannot_be_used_to_inject_a_query(repo): """The reading id is coerced before it reaches Mongo.