fix(advice): a race in the shared repository, and the prompt naming the wrong reading
Two findings from the re-review of PR #14. A race that recreated the leak ------------------------------- `get_shared_repository()` did an unsynchronised check-then-set on a module global. Eight concurrent advice threads on a cold process each saw None and each built their own repository - so eight MongoClients and twenty-five threads. The fix that was meant to remove the leak reintroduced it, just bounded by a race instead of by the number of readings. Measured before: 8 repositories built, 25 threads. Measured after: 1 repository built, 4 threads. There is a lock, and the check is repeated inside it. The prompt was about the wrong reading -------------------------------------- The advice targeted the right reading, but the PROMPT said "give advice for the most recent reading" - and when the brewer corrects an OLDER reading, the most recent one in the history is a different reading. So the model was asked about the wrong number while the advice was filed against the right one. The reading is now named in the instruction (with its date and gravity), marked in the reading list as the one to advise on, and the system prompt says to advise about the marked reading because it may not be the most recent one. A corrected reading is also flagged as such, so the model knows it may differ from what it said before. 167 tests (up from 164).
This commit is contained in:
+40
-10
@@ -67,6 +67,9 @@ temperature?
|
|||||||
- Should they rack, add more hops, raise the temperature, or wait?
|
- Should they rack, add more hops, raise the temperature, or wait?
|
||||||
|
|
||||||
Rules:
|
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 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.
|
- 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.
|
- 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:
|
def _reading_lines(readings: list[dict], subject_id: str | None = None) -> str:
|
||||||
"""The readings, as the compact list the model reads."""
|
"""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 = []
|
lines = []
|
||||||
for r in readings:
|
for r in readings:
|
||||||
parts = [f"- {r.get('date')}: gravity {r.get('sg_decimal')} ({r.get('point_scale')})"]
|
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']}")
|
parts.append(f"brewer's comment: {r['comment']}")
|
||||||
if r.get("final"):
|
if r.get("final"):
|
||||||
parts.append("(marked as the FINAL reading)")
|
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))
|
lines.append(" ".join(parts))
|
||||||
return "\n".join(lines)
|
return "\n".join(lines)
|
||||||
|
|
||||||
|
|
||||||
def build_prompt(run: dict, reading: dict, history: list[dict]) -> str:
|
def build_prompt(run: dict, reading: dict, history: list[dict],
|
||||||
"""The user-message content for one observation."""
|
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"
|
name = run.get("canonical_name") or run.get("name") or "this brew"
|
||||||
og = run.get("og_decimal")
|
og = run.get("og_decimal")
|
||||||
if og is None and run.get("og_points"):
|
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("")
|
||||||
parts.append("Readings so far (oldest first):")
|
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("")
|
||||||
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)
|
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
|
# allowance and the reply came back empty - which looked like a
|
||||||
# broken model rather than a budget problem. 2000 leaves room for
|
# broken model rather than a budget problem. 2000 leaves room for
|
||||||
# the reasoning AND the advice.
|
# the reasoning AND the advice.
|
||||||
"max_tokens": 2000,
|
"max_tokens": 3000,
|
||||||
"stream": False,
|
"stream": False,
|
||||||
}).encode("utf-8")
|
}).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], *,
|
def generate_advice(run: dict, reading: dict, history: list[dict], *,
|
||||||
|
corrected: bool = False,
|
||||||
url: str | None = None, model: str | None = None,
|
url: str | None = None, model: str | None = None,
|
||||||
timeout: float | None = None) -> tuple[bool, str]:
|
timeout: float | None = None) -> tuple[bool, str]:
|
||||||
"""Generate the advice for one reading.
|
"""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
|
short note explaining what went wrong, so the reading carries a
|
||||||
visible reason instead of silently having no advice.
|
visible reason instead of silently having no advice.
|
||||||
"""
|
"""
|
||||||
prompt = build_prompt(run, reading, history)
|
prompt = build_prompt(run, reading, history, corrected=corrected)
|
||||||
try:
|
try:
|
||||||
return True, call_ollama(prompt, url=url, model=model, timeout=timeout)
|
return True, call_ollama(prompt, url=url, model=model, timeout=timeout)
|
||||||
except urllib.error.HTTPError as exc:
|
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,
|
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.
|
"""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
|
*writer* is called as ``writer(reading_id, ok, text)``. It runs in a
|
||||||
background thread, so it must not touch the request context - the
|
background thread, so it must not touch the request context - the
|
||||||
caller is responsible for giving it a way to reach Mongo.
|
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:
|
try:
|
||||||
writer(reading.get("_id"), ok, text)
|
writer(reading.get("_id"), ok, text)
|
||||||
except Exception: # noqa: BLE001 - never let the thread die noisily
|
except Exception: # noqa: BLE001 - never let the thread die noisily
|
||||||
|
|||||||
@@ -19,6 +19,7 @@ from __future__ import annotations
|
|||||||
|
|
||||||
import logging
|
import logging
|
||||||
import os
|
import os
|
||||||
|
import threading
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
from pymongo import MongoClient
|
from pymongo import MongoClient
|
||||||
@@ -380,11 +381,28 @@ def _as_id(value: object) -> str:
|
|||||||
#: repository does too.
|
#: repository does too.
|
||||||
_shared_repository: "BeerRepository | None" = None
|
_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":
|
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
|
global _shared_repository
|
||||||
|
|
||||||
|
if _shared_repository is None:
|
||||||
|
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:
|
if _shared_repository is None:
|
||||||
_shared_repository = BeerRepository()
|
_shared_repository = BeerRepository()
|
||||||
return _shared_repository
|
return _shared_repository
|
||||||
|
|||||||
@@ -53,7 +53,7 @@ def advice_enabled() -> bool:
|
|||||||
return os.getenv("LLM_ADVICE", "1").strip().lower() not in {"0", "false", "no"}
|
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.
|
"""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
|
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(
|
thread = threading.Thread(
|
||||||
target=brew_advice.generate_advice_in_background,
|
target=brew_advice.generate_advice_in_background,
|
||||||
args=(run_info, latest, readings, write),
|
args=(run_info, latest, readings, write, corrected),
|
||||||
daemon=True,
|
daemon=True,
|
||||||
)
|
)
|
||||||
thread.start()
|
thread.start()
|
||||||
@@ -208,6 +208,14 @@ def index():
|
|||||||
"warning",
|
"warning",
|
||||||
)
|
)
|
||||||
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(
|
reading_id = repo().upsert_reading(
|
||||||
beer_run_id=run_id,
|
beer_run_id=run_id,
|
||||||
beer_type=(form.beer_run_type.data or "").strip(),
|
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
|
# The advice is generated in the background: recording a reading
|
||||||
# must not block on the model. The page shows the reading straight
|
# must not block on the model. The page shows the reading straight
|
||||||
# away and the advice is there on a later visit.
|
# 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"))
|
return redirect(url_for("brews.index"))
|
||||||
|
|
||||||
runs = repo().runs()
|
runs = repo().runs()
|
||||||
|
|||||||
@@ -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
|
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():
|
def test_the_prompt_marks_the_final_reading():
|
||||||
run = {"canonical_name": "Ale"}
|
run = {"canonical_name": "Ale"}
|
||||||
reading = {"date": "2026-01-20", "sg_decimal": 1.010, "point_scale": "1010 points",
|
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 = {}
|
captured = {}
|
||||||
|
|
||||||
def fake_generate(run, reading, history, writer=None):
|
def fake_generate(run, reading, history, writer=None, corrected=False):
|
||||||
captured["reading"] = reading
|
captured["reading"] = reading
|
||||||
captured["history"] = history
|
captured["history"] = history
|
||||||
|
captured["corrected"] = corrected
|
||||||
return True, "THE ADVICE"
|
return True, "THE ADVICE"
|
||||||
|
|
||||||
original = brews_module.brew_advice.generate_advice_in_background
|
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):
|
def test_the_reading_id_cannot_be_used_to_inject_a_query(repo):
|
||||||
"""The reading id is coerced before it reaches Mongo.
|
"""The reading id is coerced before it reaches Mongo.
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user