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).
This commit is contained in:
@@ -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:
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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")
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
|
||||
Reference in New Issue
Block a user