diff --git a/codec_premise_check.py b/codec_premise_check.py new file mode 100644 index 0000000..c5c856e --- /dev/null +++ b/codec_premise_check.py @@ -0,0 +1,186 @@ +"""Premise checking — verify attributions the USER supplies, not only CODEC's claims. + +codec_claim_check is directional: it validates assistant → user. This module +covers the other direction, and it exists because of a real miss. + +The incident (2026-07-24). The user's LinkedIn post said: + + "The most useful reply I got was from SOMEONE WHO ends each session by + listing which of his standing rules the model actually used…" + +A correspondent then wrote back: + + "YOU MENTIONED YOU end each session by listing which standing rules + actually fired…" + +The practice belonged to a commenter; the question attributed it to the user. +Both texts were in context. Neither CODEC nor the assistant flagged it — the +answer engaged with the premise and reinforced it. That is a worse failure than +invention: nobody made anything up, a false premise simply propagated as fact +through the answer, the user's reply, and onward to a third party. + +SCOPE — deliberately tiny +------------------------- +Only attributions with an artifact IN CONTEXT are checkable: the source text is +right there, so the discrepancy is a fact about two strings, not a judgement. + +Everything else — premises about the world, the user's business, their history — +has nothing to check against. A general premise-checker would just relocate the +fabrication one level up, with CODEC confidently "correcting" the user from +nothing. That is strictly worse than the gap it closes, so it is not built. + +TONE — a question, never a correction +------------------------------------- +All this can prove is that two passages disagree. It cannot know which is right; +the user may be paraphrasing loosely, or the commenter may have adopted the +practice since. So the output asks. A confident correction here would be the +same overreach this codebase exists to refuse, aimed at the user instead. + +Design bias, as everywhere in this area: FALSE NEGATIVES OVER FALSE POSITIVES. +A false positive tells the user they are wrong about their own life. The bar is +four independent conditions, all of which must hold. +""" +from __future__ import annotations + +import os +import re +from typing import List, NamedTuple + + +class PremiseFlag(NamedTuple): + practice: str # the practice being attributed, as the source words it + credited_to: str # who the source credits it to ("someone", "a commenter") + quote_source: str # the source sentence + quote_claim: str # the sentence attributing it to the user + + +# ── Attribution patterns ────────────────────────────────────────────────────── +# A source crediting a practice to a THIRD PARTY. +_THIRD_PARTY = re.compile( + r"\b(?:from|by)\s+" + r"(someone|somebody|a\s+(?:commenter|reader|colleague|friend|founder|dev(?:eloper)?|" + r"engineer|user|guy|person|reply|responder))\s+" + r"who\s+([^.!?\n]{15,200})", + re.I) + +# A message crediting a practice to the USER ("you"). +_SECOND_PERSON = re.compile( + r"\byou\s+(?:mentioned|said|noted|wrote|described|explained|told me)\s+" + r"(?:that\s+)?you\s+([^.!?\n]{15,200})", + re.I) + +_STOPWORDS = frozenset(""" +the a an of to by which who whom whose that this these those then than and or but +is are was were be been being am i you he she it we they them him her his their my +your our its as at in on for with from into over under about after before each any +some all not no do does did done have has had can could will would should may might +""".split()) + +_MIN_OVERLAP = 0.55 # containment of the shorter phrase's content words +_MIN_WORDS = 4 # below this, overlap is noise + + +def _content_words(text: str) -> set: + return {w for w in re.findall(r"[a-z]+", (text or "").lower()) + if w not in _STOPWORDS and len(w) > 2} + + +def _containment(a: set, b: set) -> float: + """Shared fraction of the SHORTER phrase — a paraphrase drops words, so + symmetric Jaccard would under-score the exact case this exists to catch.""" + if not a or not b: + return 0.0 + return len(a & b) / min(len(a), len(b)) + + +def _sentence_around(text: str, index: int) -> str: + start = max(text.rfind(".", 0, index), text.rfind("\n", 0, index)) + 1 + end = min([x for x in (text.find(".", index), text.find("\n", index)) if x != -1] + or [len(text)]) + return text[start:end].strip()[:200] + + +def _enabled() -> bool: + return (os.environ.get("PREMISE_CHECK_ENABLED") or "true").strip().lower() \ + not in ("false", "0", "no", "off") + + +def find_misattributions(texts: List[str]) -> List[PremiseFlag]: + """Practices a source credits to a third party but a later message credits + to the user. `texts` is every user-supplied string in the conversation — + pasted documents and messages alike, since the contradiction is usually + inside a single pasted thread. + + Returns [] unless all four conditions hold: + 1. a third-party attribution exists in context + 2. a second-person attribution exists in context + 3. both describe enough content to compare (>= _MIN_WORDS) + 4. they overlap above _MIN_OVERLAP + """ + if not _enabled() or not texts: + return [] + pooled = "\n".join(t for t in texts if t) + if not pooled: + return [] + + third = [(m.group(1).strip(), m.group(2).strip(), m.start()) + for m in _THIRD_PARTY.finditer(pooled)] + if not third: + return [] + second = [(m.group(1).strip(), m.start()) for m in _SECOND_PERSON.finditer(pooled)] + if not second: + return [] + + flags: List[PremiseFlag] = [] + for who, practice, t_idx in third: + pw = _content_words(practice) + if len(pw) < _MIN_WORDS: + continue + for claimed, s_idx in second: + cw = _content_words(claimed) + if len(cw) < _MIN_WORDS: + continue + if _containment(pw, cw) >= _MIN_OVERLAP: + flags.append(PremiseFlag( + practice=practice[:160], + credited_to=who.lower(), + quote_source=_sentence_around(pooled, t_idx), + quote_claim=_sentence_around(pooled, s_idx), + )) + break # one flag per third-party attribution + return flags + + +def premise_note(flags: List[PremiseFlag]) -> str: + """The note appended to a reply. Empty when there is nothing to raise. + + Phrased as a question on purpose — all that is provable here is that two + passages disagree, not which of them is right. + """ + if not flags: + return "" + f = flags[0] + # "someone" / "somebody" read badly with a possessive, so vary the closer. + anon = f.credited_to in ("someone", "somebody") + whose = "theirs" if anon else f"{f.credited_to}'s" + return ( + "\n\n---\n" + f"**Worth checking before you reply.** This is being treated as your own " + f"practice, but the text you pasted credits it to {f.credited_to} else:\n\n" + f"> {f.quote_source}\n\n" + f"vs.\n\n" + f"> {f.quote_claim}\n\n" + f"I can only tell you the two disagree, not which is right — you may be " + f"paraphrasing, or they may have picked it up since. But if it was {whose}, " + f"answering as though it were yours takes back credit you already gave away." + ) if anon else ( + "\n\n---\n" + f"**Worth checking before you reply.** This is being treated as your own " + f"practice, but the text you pasted credits it to {f.credited_to}:\n\n" + f"> {f.quote_source}\n\n" + f"vs.\n\n" + f"> {f.quote_claim}\n\n" + f"I can only tell you the two disagree, not which is right — you may be " + f"paraphrasing, or they may have picked it up since. But if it was {whose}, " + f"answering as though it were yours takes back credit you already gave away." + ) diff --git a/routes/chat.py b/routes/chat.py index ae66130..674f2bd 100644 --- a/routes/chat.py +++ b/routes/chat.py @@ -1280,6 +1280,21 @@ def _resolve_skill_tag(raw_tag): ) except Exception as e: log.warning(f"[Chat] claim check skipped (stream): {e}") + try: + import codec_premise_check + _pf = codec_premise_check.find_misattributions( + [m.get("content", "") for m in messages + if m.get("role") == "user" and isinstance(m.get("content"), str)]) + if _pf: + yield f"data: {json.dumps({'token': codec_premise_check.premise_note(_pf)})}\n\n" + log_event( + "premise_misattribution_flagged", source="codec-dashboard", + message="user attribution contradicts a pasted source", + level="warning", outcome="warning", + extra={"credited_to": _pf[0].credited_to, "transport": "stream"}, + ) + except Exception as e: + log.warning(f"[Chat] premise check skipped (stream): {e}") yield "data: [DONE]\n\n" except Exception as e: yield f"data: {json.dumps({'error': str(e)})}\n\n" @@ -1363,6 +1378,25 @@ def _resolve_skill_tag(raw_tag): except Exception as e: log.warning(f"[Chat] claim check skipped: {e}") + # Premise check: an attribution the USER supplied that a source they + # pasted contradicts. The other direction from claim-check — scope is + # deliberately tiny, see codec_premise_check. + try: + import codec_premise_check + _pf = codec_premise_check.find_misattributions( + [m.get("content", "") for m in messages + if m.get("role") == "user" and isinstance(m.get("content"), str)]) + if _pf: + answer += codec_premise_check.premise_note(_pf) + log_event( + "premise_misattribution_flagged", source="codec-dashboard", + message="user attribution contradicts a pasted source", + level="warning", outcome="warning", + extra={"credited_to": _pf[0].credited_to}, + ) + except Exception as e: + log.warning(f"[Chat] premise check skipped: {e}") + return {"response": answer, "model": model} except Exception as e: return JSONResponse({"error": str(e)}, status_code=500) diff --git a/tests/test_premise_check.py b/tests/test_premise_check.py new file mode 100644 index 0000000..cda9181 --- /dev/null +++ b/tests/test_premise_check.py @@ -0,0 +1,127 @@ +"""Premise checking — the OTHER direction from claim-check. + +The incident (2026-07-24): the user's LinkedIn post credited a practice to a +commenter ("the most useful reply I got was from SOMEONE WHO ends each session +by listing…"). A correspondent then wrote back attributing it to the user ("YOU +MENTIONED YOU end each session by listing…"). Both texts were in context. The +assistant engaged with the premise and reinforced it — nobody invented anything, +a false premise simply propagated as fact. + +The design bias is the strongest in the codebase, because a false positive here +tells the user they are wrong about their own life. Most of these tests are +false-positive guards. +""" +from __future__ import annotations + +import sys +from pathlib import Path + +import pytest + +_REPO = Path(__file__).resolve().parents[1] +if str(_REPO) not in sys.path: + sys.path.insert(0, str(_REPO)) + +import codec_premise_check as pc # noqa: E402 + + +POST = ("I posted the full set to a developer community last week. The most useful " + "reply I got was from someone who ends each session by listing which of his " + "standing rules the model actually used, then deletes the ones that never fire.") +QUESTION = ("You mentioned you end each session by listing which standing rules " + "actually fired, then delete the ones that never trigger. In that cleanup " + "process, was there ever a rule you had to delete?") + + +# ── the real incident ───────────────────────────────────────────────────────── +def test_the_actual_misattribution_is_caught(): + flags = pc.find_misattributions([POST, QUESTION]) + assert len(flags) == 1 + assert flags[0].credited_to == "someone" + note = pc.premise_note(flags) + assert "credits it to someone" in note + assert "not which is right" in note, "must not assert who is correct" + + +def test_note_is_a_question_not_a_verdict(): + """All this can prove is that two passages disagree.""" + note = pc.premise_note(pc.find_misattributions([POST, QUESTION])) + for overclaim in ("you are wrong", "incorrect", "that is false", "you did not"): + assert overclaim not in note.lower() + assert "worth checking" in note.lower() + + +def test_it_works_across_separate_messages(): + """The source and the claim need not be in the same message.""" + assert pc.find_misattributions([POST, "ok thanks", QUESTION]) + + +def test_it_works_inside_one_pasted_thread(): + """The usual shape: a whole conversation pasted as one message.""" + assert pc.find_misattributions([POST + "\n\n" + QUESTION]) + + +# ── FALSE-POSITIVE GUARDS — the tests that matter most ──────────────────────── +@pytest.mark.parametrize("texts", [ + # third-party credit + an UNRELATED you-statement + ["I got a tip from someone who runs their tests in a container before deploy.", + "You mentioned you prefer working late at night."], + # you-statement with no source to contradict it + ["You mentioned you always verify the service is running before any step."], + # third-party credit with no you-claim + ["The best reply was from someone who deletes rules that never fire."], + # genuinely the user's own practice, stated by them + ["I end each session by listing which rules fired and deleting dead ones.", + "You mentioned you end each session by listing which rules fired."], + # a colleague quoted on a different subject + ["A colleague who reviews every PR twice told me it halves defects.", + "You said you ship on Fridays."], + # phrases too thin to compare + ["a tip from someone who tests more", "You mentioned you test more"], + # ordinary conversation + ["What's the weather?", "Can you summarise this document?"], + ["", ""], + [], +]) +def test_no_false_positive(texts): + assert pc.find_misattributions(texts) == [], texts + + +def test_partial_topic_overlap_is_not_enough(): + """Sharing a couple of words must not trip it — only a real paraphrase.""" + a = "The idea came from someone who writes standing rules for every project." + b = "You mentioned you write documentation for every project." + assert pc.find_misattributions([a, b]) == [] + + +# ── operational ─────────────────────────────────────────────────────────────── +def test_kill_switch(monkeypatch): + monkeypatch.setenv("PREMISE_CHECK_ENABLED", "false") + assert pc.find_misattributions([POST, QUESTION]) == [] + + +def test_empty_flags_produce_no_note(): + assert pc.premise_note([]) == "" + + +def test_named_third_party_reads_naturally(): + """'a colleague' should possessivise; 'someone' should not.""" + a = ("The approach came from a colleague who batches every migration into a " + "single reviewed transaction before it touches production data.") + b = ("You mentioned you batch every migration into a single reviewed " + "transaction before it touches production data.") + flags = pc.find_misattributions([a, b]) + assert flags, "a named third party must still be caught" + note = pc.premise_note(flags) + assert "colleague's" in note and "someone else" not in note + + +def test_wired_into_both_chat_paths(): + """Guard the wiring against a refactor silently dropping it.""" + src = (_REPO / "routes" / "chat.py").read_text() + assert src.count("codec_premise_check") >= 4, "must be wired in stream AND non-stream" + gen = src[src.index("def _stream_gen("):] + gen = gen[:gen.index("from starlette.responses import StreamingResponse")] + assert "codec_premise_check" in gen, "streaming path must run the premise check" + assert gen.index("codec_premise_check") < gen.index('"data: [DONE]'), \ + "the check must run before [DONE]"