diff --git a/carwatch/agent.py b/carwatch/agent.py index 9f3712e..db0cad4 100644 --- a/carwatch/agent.py +++ b/carwatch/agent.py @@ -459,7 +459,10 @@ def _think(question: str, asker: str) -> str: voicestate.record_answer_s(time.time() - t0, model=_serving_model_name()) if line_buf.strip(): print(f" ... {line_buf.strip()}", flush=True) - answer = "".join(parts).strip() + from carwatch.grounding import strip_scaffold + # The driver gets the answer, not the worksheet (#52): some models open + # by echoing the prompt's own KNOWN FACTS block. + answer = strip_scaffold("".join(parts).strip()) # Hand the dash the finished text; the voice path overrides this with # "speaking" right after, other surfaces are simply done. voicestate.set_state("idle", answer=answer[:400]) diff --git a/carwatch/grounding.py b/carwatch/grounding.py index eadf9cd..aeb8229 100644 --- a/carwatch/grounding.py +++ b/carwatch/grounding.py @@ -14,9 +14,10 @@ `cannot_sense` and into `facts`, and the car earns the right to talk about them. """ - from __future__ import annotations +import re + # Default identity: the GLE. Overridden per car via build_system_prompt's # identity arg (fed from the config's `car` block) so the same code serves # @gle in Berlin and @eclass in Helsinki without edits (the Helsinki move). @@ -38,6 +39,51 @@ Answer in the LANGUAGE the question was asked in: Finnish gets Finnish, English gets English. Voice transcripts may be imperfect Finnish; answer the likely intent in Finnish rather than declaring the message unparseable.""" +# The headings this module writes INTO the prompt. A model that echoes its +# worksheet starts its reply with one of them, so they are also the markers +# for stripping it back out (#52: an /ask answer opened with the verbatim +# "- KNOWN FACTS: OBD is running but cable NOT U..."). Defined once so the +# prompt and the stripper can never drift apart. +FACTS_HEADING = "KNOWN FACTS" +CANNOT_HEADING = "YOU CANNOT SENSE" +RULES_HEADING = "STRICT GROUNDING RULES" +_SCAFFOLD_MARKERS = (FACTS_HEADING, CANNOT_HEADING, RULES_HEADING) + +_THINK_BLOCK = re.compile(r"<(think|thinking|reasoning)>.*?", + re.IGNORECASE | re.DOTALL) +_BULLET = re.compile(r"^\s*(?:[-*\u2022]|\d+[.)])\s+") + + +def strip_scaffold(text: str) -> str: + """Give the driver the answer, not the worksheet. + + Removes any reasoning block, then drops LEADING lines that are the + prompt's own scaffold plus the fact bullets trailing them. Conservative + on purpose: it only strips from the START, it matches the headings in + their prompt casing so ordinary prose cannot trip it, and if stripping + would leave nothing it returns the original. A scaffolded answer is bad; + an empty one is worse. + """ + if not text: + return text + cleaned = _THINK_BLOCK.sub("", text).strip() + lines = cleaned.splitlines() + i, dropping = 0, False + while i < len(lines): + bare = _BULLET.sub("", lines[i]).strip().lstrip("#").strip() + if any(bare.startswith(m) or bare[:40].find(m) >= 0 + for m in _SCAFFOLD_MARKERS): + dropping = True + i += 1 + continue + if dropping and (not bare or _BULLET.match(lines[i])): + i += 1 # the fact bullets that follow a heading + continue + break + out = "\n".join(lines[i:]).strip() + return out or cleaned or text + + def build_system_prompt( facts: dict[str, str] | None = None, cannot_sense: list[str] | None = None, @@ -52,7 +98,8 @@ def build_system_prompt( "a Raspberry Pi 5 named Vadelma running a language model fully offline, no internet") rules = RULES.format(identity=identity or DEFAULT_IDENTITY) - lines = [rules, "", "KNOWN FACTS (the only current state you may assert):"] + lines = [rules, "", + f"{FACTS_HEADING} (the only current state you may assert):"] # Car state first, the Pi's own vitals last: models lead with whatever # is listed first, and a status answer that opens with CPU fans instead # of fuel and tyres flattens the whole point of a car that talks @@ -70,7 +117,7 @@ def _is_computer(key): lines.append(f"- {k}: {v}") if cannot_sense: - lines += ["", "YOU CANNOT SENSE THESE AT ALL RIGHT NOW (say so if asked):"] + lines += ["", f"{CANNOT_HEADING} THESE AT ALL RIGHT NOW (say so if asked):"] lines += [f"- {item}" for item in cannot_sense] if manual_excerpts: diff --git a/carwatch/webchat.py b/carwatch/webchat.py index 3c2063f..91f11cd 100644 --- a/carwatch/webchat.py +++ b/carwatch/webchat.py @@ -1229,7 +1229,8 @@ def answer(question: str, use_manual: bool = True) -> str: except Exception: _vs.set_state("idle", note="answer failed - brain unreachable?") raise - _out = (msg.get("content") or "").strip() + from carwatch.grounding import strip_scaffold + _out = strip_scaffold((msg.get("content") or "").strip()) _vs.set_state("idle", answer=_out[:400]) return _out or "[the model spent its budget thinking and did not answer]" diff --git a/tests/test_scaffold_leak.py b/tests/test_scaffold_leak.py new file mode 100644 index 0000000..5a0f1b6 --- /dev/null +++ b/tests/test_scaffold_leak.py @@ -0,0 +1,62 @@ +"""#52: /ask answers began with the prompt's own worksheet. + +Observed on VTA 14 Sep: the reply opened with the verbatim +"- KNOWN FACTS: OBD is running but cable NOT U..." before the actual answer. +The driver should get the answer, not the scaffold. +""" +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from carwatch.grounding import ( # noqa: E402 + CANNOT_HEADING, FACTS_HEADING, build_system_prompt, strip_scaffold, +) + + +class ScaffoldLeak(unittest.TestCase): + + def test_the_reported_leak_is_removed(self): + out = strip_scaffold( + "- KNOWN FACTS: OBD is running but cable NOT UP\n" + "- fuel: 58%\n\nYour tyres look fine, Petrus.") + self.assertEqual(out, "Your tyres look fine, Petrus.") + + def test_a_real_echoed_heading_and_its_bullets_go(self): + # Build the scaffold from the prompt itself, so this test tracks the + # real wording rather than a copy that can drift. + p = build_system_prompt(facts={"fuel": "58%", "tyres": "2.4 bar"}) + lines = p.splitlines() + start = next(i for i, l in enumerate(lines) if l.startswith(FACTS_HEADING)) + echoed = "\n".join(lines[start:start + 4]) + "\n\nHalf a tank, Petrus." + self.assertEqual(strip_scaffold(echoed), "Half a tank, Petrus.") + + def test_reasoning_blocks_are_removed(self): + self.assertEqual( + strip_scaffold("weighing it upYour tyres look fine."), + "Your tyres look fine.") + + def test_ordinary_prose_is_untouched(self): + for s in ("I cannot sense that yet.", + "I only state what appears in KNOWN FACTS, and that is half a tank.", + "Your coolant is 90 degrees."): + self.assertEqual(strip_scaffold(s), s) + + def test_never_returns_empty(self): + # Over-stripping must not silence the car. A scaffolded answer is bad; + # no answer at all is worse. + only = "- " + FACTS_HEADING + ": only this" + self.assertTrue(strip_scaffold(only)) + self.assertEqual(strip_scaffold(""), "") + + def test_prompt_and_stripper_share_the_headings(self): + # If someone reworded the prompt, the stripper must move with it. + p = build_system_prompt(facts={"fuel": "58%"}, + cannot_sense=["outside temperature"]) + self.assertIn(FACTS_HEADING, p) + self.assertIn(CANNOT_HEADING, p) + + +if __name__ == "__main__": + unittest.main()