Anti-hallucination: 5-layer grounding layer (quote verify, trigger va…

backend/app/engines/adversary.py

@@ -3,6 +3,7 @@
 """
 from __future__ import annotations
 from .. import llm, config
+from . import grounding
 
 PERSONA_DESC = {
     "Hardball":   "Direct, aggressive. Uses blunt pressure lines ('standard clause', 'non-negotiable'). Rarely volunteers concessions. Every position is stated as final until genuinely earned.",
@@ -181,17 +182,38 @@ def respond(session: dict, user_msg: str) -> dict:
     history.append({"role": "user", "content": user_msg})
 
     out      = llm.chat_json(history, model=config.ADVERSARY_MODEL, temperature=0.75, max_tokens=800)
+    out      = grounding.validate_adversary_output(out)        # schema guard
     reply    = out.get("reply", "")
     conceded = bool(out.get("conceded"))
     gate     = None
 
     if conceded:
-        gate = _referee(sc, user_msg, out.get("trigger_claimed", ""), difficulty)
-        if not gate.get("earned"):
+        claimed  = out.get("trigger_claimed", "")
+        triggers = sc["adversary"].get("concession_triggers", [])
+
+        # trigger validation: claimed trigger must match a known trigger
+        if not grounding.trigger_is_valid(claimed, triggers):
             conceded = False
             reply = (reply.rstrip() +
-                     "\n\n[That said — you haven't given me a compelling enough reason to move on this. My position stands.]")
+                     "\n\n[That said — I need a stronger basis than that. My position stands.]")
             out["concession_detail"] = ""
+            gate = {"earned": False, "reason": "Claimed trigger did not match any valid concession condition.",
+                    "_grounding": "trigger_validation_failed"}
+        else:
+            # self-consistency referee: 3 independent calls, majority vote
+            gate = grounding.majority_vote(
+                _referee,
+                args=(sc, user_msg, claimed, difficulty),
+                kwargs={},
+                n=3,
+                key="earned",
+            )
+            gate = grounding.validate_referee_output(gate)
+            if not gate.get("earned"):
+                conceded = False
+                reply = (reply.rstrip() +
+                         "\n\n[That said — you haven't given me a compelling enough reason to move on this. My position stands.]")
+                out["concession_detail"] = ""
 
     session["turn"] += 1
     session["transcript"].append({"role": "user", "content": user_msg})

backend/app/engines/grounding.py

@@ -0,0 +1,206 @@
+"""Anti-hallucination grounding layer for Whetstone.
+
+Five mechanisms:
+  1. Quote verification   — evidence_quote must appear in the actual transcript
+  2. Trigger validation   — claimed trigger must match a known trigger (fuzzy)
+  3. Self-consistency     — referee called N times; majority vote wins
+  4. Score sanity         — numeric scores clamped to valid ranges
+  5. Schema guard         — required fields and types enforced on all JSON output
+"""
+from __future__ import annotations
+import re
+from difflib import SequenceMatcher
+from typing import Any, Callable
+
+
+# ── 1. Quote verification ─────────────────────────────────────────────────────
+
+def _normalise(s: str) -> str:
+    return re.sub(r"\s+", " ", s.lower().strip())
+
+
+def quote_in_transcript(quote: str, transcript: list[dict], threshold: float = 0.65) -> bool:
+    """Return True if quote appears verbatim or near-verbatim in any transcript turn."""
+    if not quote or not quote.strip():
+        return True  # empty quote — nothing to verify
+    q = _normalise(quote)
+    for turn in transcript:
+        body = _normalise(turn.get("content", ""))
+        # exact substring
+        if q in body:
+            return True
+        # sliding-window fuzzy match (handles minor whitespace / ellipsis diffs)
+        words_q = q.split()
+        words_b = body.split()
+        window = len(words_q)
+        for i in range(max(1, len(words_b) - window + 1)):
+            chunk = " ".join(words_b[i : i + window])
+            ratio = SequenceMatcher(None, q, chunk).ratio()
+            if ratio >= threshold:
+                return True
+    return False
+
+
+def verify_scoring_quotes(result: dict, transcript: list[dict]) -> dict:
+    """
+    Check every rubric_result evidence_quote against the actual transcript.
+    If a quote cannot be found, mark it as unverified and zero the score
+    to prevent evidence-hallucinated verdicts from standing.
+    """
+    rubric_results = result.get("rubric_results", [])
+    any_fixed = False
+    for item in rubric_results:
+        quote = item.get("evidence_quote", "")
+        if quote and not quote_in_transcript(quote, transcript):
+            item["_quote_unverified"] = True
+            item["comment"] = (
+                f"[GROUNDING: quote not found in transcript] {item.get('comment', '')}"
+            )
+            # downgrade verdict — unverified evidence ≠ earned point
+            if item.get("verdict") in ("MET", "1", "2"):
+                item["verdict"] = "UNMET"
+                item["_auto_corrected"] = True
+                any_fixed = True
+    if any_fixed:
+        result["_grounding_corrections"] = True
+    return result
+
+
+# ── 2. Trigger validation ─────────────────────────────────────────────────────
+
+def trigger_is_valid(claimed: str, valid_triggers: list[str], threshold: float = 0.45) -> bool:
+    """
+    Return True if claimed trigger is semantically related to any valid trigger.
+    Uses token-overlap ratio — good enough for 1-2 sentence trigger descriptions.
+    """
+    if not claimed:
+        return False
+    c = _normalise(claimed)
+    for t in valid_triggers:
+        t_norm = _normalise(t)
+        ratio = SequenceMatcher(None, c, t_norm).ratio()
+        if ratio >= threshold:
+            return True
+        # keyword overlap fallback
+        c_words = set(c.split()) - {"the", "a", "an", "of", "in", "and", "or", "to", "is"}
+        t_words = set(t_norm.split()) - {"the", "a", "an", "of", "in", "and", "or", "to", "is"}
+        if c_words and t_words:
+            overlap = len(c_words & t_words) / min(len(c_words), len(t_words))
+            if overlap >= 0.40:
+                return True
+    return False
+
+
+# ── 3. Self-consistency referee ───────────────────────────────────────────────
+
+def majority_vote(
+    fn: Callable[..., dict],
+    args: tuple,
+    kwargs: dict,
+    n: int = 3,
+    key: str = "earned",
+) -> dict:
+    """
+    Call fn(*args, **kwargs) n times and return the majority result.
+    Ties default to the conservative answer (False for 'earned').
+    """
+    results: list[dict] = []
+    for _ in range(n):
+        try:
+            results.append(fn(*args, **kwargs))
+        except Exception:  # noqa: BLE001
+            results.append({key: False, "reason": "call failed"})
+
+    votes_true  = sum(1 for r in results if r.get(key) is True)
+    votes_false = n - votes_true
+    winner = votes_true > votes_false  # ties → False (conservative)
+
+    # Return the result whose vote matched the majority
+    matching = [r for r in results if r.get(key) is winner]
+    best = matching[0] if matching else results[0]
+    best["_consistency_votes"] = {"true": votes_true, "false": votes_false, "n": n}
+    return best
+
+
+# ── 4. Score sanity ───────────────────────────────────────────────────────────
+
+_DIMENSION_NAMES = {
+    "Position Defense", "Concession Strategy",
+    "Anchoring & Offers", "Pressure Resistance", "Legal Reasoning",
+}
+
+
+def clamp_scores(result: dict) -> dict:
+    """Ensure dimension scores are non-negative integers within [0, max]."""
+    dim_scores = result.get("dimension_scores", {})
+    for dim, data in dim_scores.items():
+        if not isinstance(data, dict):
+            continue
+        raw_score = data.get("score", 0)
+        raw_max   = data.get("max",   0)
+        try:
+            score = int(round(float(raw_score)))
+            max_  = int(round(float(raw_max)))
+        except (TypeError, ValueError):
+            score, max_ = 0, 0
+        data["score"] = max(0, min(score, max(max_, 0)))
+        data["max"]   = max(0, max_)
+
+    # overall_score must be 0-100
+    ov = result.get("overall_score")
+    try:
+        result["overall_score"] = max(0, min(100, int(round(float(ov)))))
+    except (TypeError, ValueError):
+        result["overall_score"] = 0
+
+    return result
+
+
+# ── 5. Schema guard ───────────────────────────────────────────────────────────
+
+def _ensure_str(d: dict, key: str, default: str = "") -> None:
+    if not isinstance(d.get(key), str):
+        d[key] = default
+
+
+def _ensure_bool(d: dict, key: str, default: bool = False) -> None:
+    v = d.get(key)
+    if not isinstance(v, bool):
+        # accept "true"/"false" strings from LLM
+        if isinstance(v, str):
+            d[key] = v.lower() in ("true", "1", "yes")
+        else:
+            d[key] = default
+
+
+def validate_adversary_output(out: dict) -> dict:
+    """Enforce schema on adversary JSON: required fields, correct types."""
+    _ensure_str(out, "reply", "[No response generated]")
+    _ensure_bool(out, "conceded", False)
+    _ensure_str(out, "concession_detail", "")
+    _ensure_str(out, "trigger_claimed", "")
+    # reply must be non-empty
+    if not out["reply"].strip():
+        out["reply"] = "[No response generated]"
+    return out
+
+
+def validate_referee_output(out: dict) -> dict:
+    """Enforce schema on referee JSON."""
+    _ensure_bool(out, "earned", False)
+    _ensure_str(out, "reason", "No reason provided.")
+    return out
+
+
+def validate_score_output(out: dict) -> dict:
+    """Enforce schema on scoring JSON and clamp numeric fields."""
+    if not isinstance(out.get("rubric_results"), list):
+        out["rubric_results"] = []
+    if not isinstance(out.get("dimension_scores"), dict):
+        out["dimension_scores"] = {d: {"score": 0, "max": 0} for d in _DIMENSION_NAMES}
+    _ensure_str(out, "summary", "Scoring unavailable.")
+    _ensure_str(out, "attribution", "")
+    if not isinstance(out.get("turning_point"), dict):
+        out["turning_point"] = {"turn": 1, "what_happened": "", "stronger_play": ""}
+    out = clamp_scores(out)
+    return out

backend/app/engines/scoring.py

@@ -3,6 +3,7 @@
 """
 from __future__ import annotations
 from .. import llm, config
+from . import grounding
 
 
 def _transcript_text(session: dict) -> str:
@@ -93,6 +94,8 @@ def score(session: dict) -> dict:
     ]
     result = llm.chat_json(msg, model=config.JUDGE_MODEL, temperature=0.15, max_tokens=3000,
                            retries=2)
+    result = grounding.validate_score_output(result)               # schema guard + score clamp
+    result = grounding.verify_scoring_quotes(result, session["transcript"])  # quote grounding
     session["finished"] = True
     session["score"] = result
     return result