#!/usr/bin/env python3
"""
Phase 3 SFT & Abstention Data Builder + Structural Speech-Act Surgeon
=====================================================================
Implements `PLAN.md` §2.2, §2.4, §2.6, and §2.7.A #27–#28 (with all fixes from
`audits/opus_audit_phase3_prep.md` and `audits/opus_audit_phase3_prep_v2.md`):
1. In-Place Sentence/Clause Speech-Act Surgery (`apply_speech_act_surgery`):
- Preserves all newlines, bullet lists, markdown tables, aligned indentation, and fenced code blocks
(`B2` & `B2'`: 0/140 fenced Gemini reference turns have their fence contents modified).
- Strips sycophantic validation openers ("You're right!", "Good catch!", "Great question!",
"Makes sense!") ONLY when not followed by a contrastive conjunction (`but`, `except`, `though`,
`however`, `yet`, `still` — `M3`).
- Strips pre-tool announcement sentences/clauses at the start or end of a turn before `<tool_call>`
(English + Japanese `。`, including "Let me check...", "Now let me...", "Allow me to...",
"And I'll...", "So I'll...", "確認します。", "見てみます。"), while preserving causal/hypothesis/colon/then
clauses ("Let me check X because Y", "I'll test this hypothesis: Y", "I'll run X first, then Y")
and never deleting clarification questions ending in `?` (`M2` & `M2'`).
- Strips trailing closing permission-solicitations ONLY when the final sentence is a genuine offer
or single-action permission question ("Do you want me to...?", "If you want, I can apply...",
"Let me know if...", "Feel free to...", "Happy to help..."), NEVER stripping `X or Y?` choice forks
("Should I write the parser in Rust or Go?", "Would you like X or Y?"), `I can't...`, `I can confirm...`,
or technical conditionals ("If you want lower prefill latency, set `--ubatch 4096`.", `M1` & `M1'`).
- Strips modern pictographic emoji (`U+1F300..U+1FAFF`), regional indicator flags (`U+1F1E6..U+1F1FF`),
skin tones, `U+FE0F`, `U+200D`, and `✅ ❌ ⚠️ ✨ ⭐ ⏰ ⌛`, absorbing only the space directly adjacent to
the removed emoji while preserving kaomoji (`✧ ♡ ♥ ✿ ☆ ♪ (◕‿◕✿) ( ; ω ; ) (´• ω •`)`) and inline code (`B3` & `B3'`).
- Identifier-safe prose lowercasing (`M4`): protects fenced/inline code, URLs, POSIX/Windows paths, CLI flags,
dotted filenames (`README.md`, `wg0.conf`), words with `>= 2` uppercase letters (`PPPoE`, `macOS`, `iOS`,
`HOME`, `CUDA_VISIBLE_DEVICES`, `MTU`, `WAL`, `WireGuard`, `MemoryHigh`, `OOMPolicy`), tokens containing
digits or `=` (`K=320`, `Qwen3.8-Flash-Next`, `Q2_0`, `IQ3_S`), and `Makefile`/`Dockerfile`.
2. Read-only style & speech-act audit on Albedo `gemini-3.8-flash` and `gemini-4-argon` assistant
turns (`PLAN.md` §2.2 L66), measuring surgery rates without exporting personal session text.
3. R-Tuning Abstention-Aware Healing Classifier (`classify_rtuning_abstention`, `PLAN.md` §2.4, `M5–M6`):
- Word-boundary matching (`(?<![a-z0-9_])...(?![a-z0-9_])` so `"42"` never matches `"1042"`), where a flat
list of strings is ALWAYS treated as any-of aliases (`["42", "forty-two"]`) and nested lists
(`[["42", "forty-two"], ["seconds", "sec"]]`) require every group to match at least one alias.
- Keeps teacher answer + `use_teacher_topk_logits=True` when pruned student samples agree (`>= 0.75`),
recording `kd_logit_extraction_order = "extract_teacher_topk_logits_after_surgery_on_target_response"`.
- Rewrites lost/inconsistent facts to a diverse pool of persona-voice abstention templates (`ABSTENTION_TEMPLATES`,
verified free of all `phase3_speech_act_ngrams` and `phase4_lexical_ftpo_ngrams`) or a silent `web_search`
tool call (`target_response = ""`) with `use_teacher_topk_logits=False`.
4. Verifies 100% rule coverage over `results/banned_llm_isms.json` `phase3_speech_act_ngrams` in `--self-test` (`M7`).
"""
import argparse
from collections import Counter
import hashlib
import json
import os
import re
import sqlite3
import sys
from pathlib import Path
from typing import Dict, List, Optional, Sequence, Tuple, Union
sys.path.insert(0, str(Path(__file__).resolve().parent))
from mine_llm_isms_from_transcripts import decode_etf_term
# 1. Sycophantic opener at the start of a reply (M3: never strip before contrastive conjunctions!)
SYCOPHANTIC_OPENER_RE = re.compile(
r"^\s*(?:(?:yep|yeah|yes|oh|ah|right)[,!\s]+)*(?:"
r"you(?:'re|\u2019re|\s+are)\s+(?:completely\s+|totally\s+|absolutely\s+|100%\s+|so\s+)?(?:right|correct|spot\s+on)"
r"|(?:good|great|nice|excellent|fair|sharp)\s+(?:catch|question|point|call|observation|eye)"
r"|spot\s+on"
r"|makes\s+(?:total\s+|complete\s+)?sense"
r"|that(?:'s|\u2019s|\s+is)\s+a\s+(?:great|good|fair|valid)\s+(?:question|point|catch)"
r"|absolutely"
r"|exactly\s+right"
r")\s*[.!?,;:\u2014\-]+\s*",
re.IGNORECASE,
)
CONTRASTIVE_START_RE = re.compile(
r"^(?:but|except|though|although|however|yet|still|wait|actually)\b",
re.IGNORECASE,
)
# 2. Pre-tool announcement clause with a causal/hypothesis/colon/then continuation (M2 & M2': keep the hypothesis/plan!)
PRE_TOOL_HYPOTHESIS_CLAUSE_RE = re.compile(
r"^\s*(?:now\s+|so\s+|and\s+|first,?\s+|ok(?:ay)?,?\s+|alright,?\s+)*"
r"(?:let\s+me\s+(?:actually\s+|just\s+|quickly\s+|go\s+ahead\s+and\s+|first\s+)*"
r"|allow\s+me\s+to\s+(?:just\s+|quickly\s+)*"
r"|i(?:'ll|\u2019ll|\s+will)\s+(?:just\s+|quickly\s+|go\s+ahead\s+and\s+|first\s+)*)"
r"(?:check|look(?:\s+at|\s+into)?|inspect|examine|read|view|verify|confirm|find|search|run|test|see|pull\s+up|grab|fetch|open|list)\b"
r"[^?:.!?\n]*?(?:,\s*(?:since|because|as|then)\b\s+|\s+\b(?:because|since)\b\s+|:\s+)(.+)$",
re.IGNORECASE,
)
# Standalone pre-tool announcement sentence (English + Japanese, M2 & M2': never delete questions with '?' or ':' clauses!)
PRE_TOOL_NARRATION_SENT_RE = re.compile(
r"^\s*(?:"
r"(?:(?:now|so|and|first|next|then|ok|okay|alright|sure)[,\s]+)*"
r"(?:"
r"(?:let\s+me|lem\s+me|allow\s+me\s+to)\s+(?:actually\s+|just\s+|quickly\s+|go\s+ahead\s+and\s+|first\s+)*"
r"(?:check|look|inspect|examine|read|view|verify|confirm|find|search|run|test|see|pull\s+up|grab|fetch|open|list|start|take\s+a\s+look)\b"
r"|i(?:'ll|\u2019ll|\s+will|\s+am\s+going\s+to|(?:'m|\u2019m)\s+going\s+to)\s+(?:actually\s+|just\s+|quickly\s+|go\s+ahead\s+and\s+|first\s+)*"
r"(?:check|look|inspect|examine|read|view|verify|confirm|find|search|run|test|see|pull\s+up|grab|fetch|open|list|take\s+a\s+look)\b"
r"|let(?:'s|\u2019s)\s+(?:just\s+|quickly\s+|first\s+)*(?:check|look\s+at|inspect|read|verify|confirm|see\s+what)\b"
r")"
r"[^?:\uff1f\n]*[.!\u3002\uff01]?"
r"|(?:\u4e86\u89e3(?:\u3067\u3059)?|\u627f\u77e5\u3057\u307e\u3057\u305f|\u308f\u304b\u308a\u307e\u3057\u305f)[\u3002\uff01]?"
r"|(?:(?:\u4e86\u89e3(?:\u3067\u3059)?|\u627f\u77e5\u3057\u307e\u3057\u305f|\u308f\u304b\u308a\u307e\u3057\u305f)[\u3002\uff01]?\s*)?"
r"(?:\u307e\u305a\u306f|\u3055\u3063\u305d\u304f)?[^\u3002\uff01\uff1f\n]{0,30}"
r"(?:\u78ba\u8a8d\u3057\u307e\u3059|\u78ba\u8a8d\u3057\u3066\u307f\u307e\u3059|\u898b\u3066\u307f\u307e\u3059|\u8abf\u3079\u3066\u307f\u307e\u3059|\u8aad\u3093\u3067\u307f\u307e\u3059|\u5b9f\u884c\u3057\u307e\u3059)[\u3002\uff01]?"
r")\s*$",
re.IGNORECASE,
)
# 3. Closing permission-solicitation / menu-offering sentence at the end of a turn (M1 & M1')
OFFER_ACTION_VERB_PAT = (
r"(?:go\s+ahead\s+and\s+|just\s+|quickly\s+)*"
r"(?:apply|run|push|commit|write|edit|update|fix|do|set(?:\s+up)?|make|create|add|remove|delete|"
r"install|start|stop|restart|build|test|patch|wire|implement|hook|draft|generate|prepare|pull|check|handle|take\s+care)"
)
CLOSING_SOLICITATION_OFFER_RE = re.compile(
rf"^\s*(?:"
# Second-person permission question ending in '?'
rf"(?:(?:so|and|now|or)[,\s]+)*(?:"
rf"(?:do\s+you\s+want|would\s+you\s+like|would\s+you\s+prefer|would\s+you\s+rather|want)\s+(?:me\s+to\b|to\b|it\b)"
rf"|shall\s+i\b"
rf"|should\s+i\s+{OFFER_ACTION_VERB_PAT}\b"
rf"|care\s+for\s+me\s+to\b"
rf"|you\s+want\s+(?:me\s+to\b|to\b|it\b)"
rf")[^?\n]*\?"
# Explicit first-person action offer ("If you want [me to / it], I can/will/we can <action_verb> ...")
rf"|(?:if\s+you\s+(?:want(?:\s+me\s+to|\s+to|\s+it)?|(?:'d|\u2019d|\s+would)\s+(?:like|rather|prefer))|whenever\s+you(?:'re|\u2019re|\s+are|\s+want\s+me\s+to))\b"
rf"[^.!?\n]*?\b(?:(?:i\s+(?:can(?!'t|\u2019t|not)|will|could)|i(?:'ll|\u2019ll)|we\s+can)\s+{OFFER_ACTION_VERB_PAT}\b|i(?:'d|\u2019d)\s+be\s+happy\s+to\b|let\s+me\s+know|just\s+say\s+the\s+word|say\s+the\s+word)[^.!?\n]*[.!?]?"
# Standalone "Let me know if..." / "Feel free to..." / "Happy to help..." / "Your call..."
rf"|(?:just\s+|please\s+)?let\s+me\s+know\s+(?:if|what|whether|how|when|whenever)\b[^.!?\n]*[.!?]?"
rf"|feel\s+free\s+to\s+(?:ask|let\s+me\s+know|reach\s+out|ping|tell\s+me)\b[^.!?\n]*[.!?]?"
rf"|(?:i(?:'m|\u2019m|\s+am)\s+)?(?:always\s+)?(?:happy|glad)\s+to\s+(?:help|assist|do\s+that|run\s+that|set\s+that\s+up|walk\s+through)\b[^.!?\n]*[.!?]?"
rf"|(?:it(?:'s|\u2019s)\s+)?your\s+call(?:\s+on\s+whether\b|\s+if\s+you\s+want\b|\s*[.!])[^.!?\n]*[.!?]?"
rf"|just\s+say\s+the\s+word\b[^.!?\n]*[.!?]?"
rf"|(?:i\s+)?(?:don(?:'|\u2019)?t\s+want\s+to|wouldn(?:'|\u2019)?t\s+want\s+to|i(?:'d|\u2019d)\s+want\s+to)\s+(?:touch|change|overwrite|modify|run)\b[^.!?\n]*?\bwithout\s+asking\b[^.!?\n]*[.!?]?"
rf")\s*$",
re.IGNORECASE,
)
# Guard against stripping genuine technical trade-off questions or 'X or Y?' design forks (M1 & M1')
GENUINE_TECH_TRADEOFF_RE = re.compile(
r"\b(?:k\s*=\s*\d+|q2_0|iq3_s|iq2_xs|gdn|qsa|moce|scr|ftpo|lora|gguf|ppl|kld|vram|nvme|pcie|tflop)\b",
re.IGNORECASE,
)
CHOICE_FORK_QUESTION_RE = re.compile(r"^\s*(?!or\b)\S[^?\n]*?\s+(?:or|vs\.?|versus)\s+(?!(?:not|anything\s+else|something\s+else|would\s+you|do\s+you|should\s+i|shall\s+i|wait|skip|leave\s+it)\b)\S[^?\n]*\?\s*$", re.IGNORECASE)
# 4. B3 & B3'-safe Unicode emoji regex: strips modern pictographic emoji (U+1F300..U+1FAFF), regional indicator flags
# (U+1F1E6..U+1F1FF), skin tones (U+1F3FB..U+1F3FF), FE0F/200D, and ✅❌⚠️✨⭐⏰⌛,
# while PRESERVING U+2600..U+27BF kaomoji decorative symbols (✧ ♡ ♥ ✿ ☆ ♪ ★ ❀ ❁ ♨)!
UNICODE_EMOJI_RE = re.compile(
r"[\U0001F1E6-\U0001F1FF\U0001F300-\U0001FAFF\U0001F3FB-\U0001F3FF\u2705\u274C\u26A0\u2728\u2B50\u23F0\u231B\uFE0F\u200D]+"
)
EMOJI_WITH_ADJACENT_SPACE_RE = re.compile(
r"([ \t]*)(" + UNICODE_EMOJI_RE.pattern + r")([ \t]*)"
)
# 5. M4 & M4'-safe identifier-preserving spans for prose lowercasing
PROTECTED_SPAN_RE = re.compile(
r"(```[\s\S]*?```"
r"|`[^`\n]+`"
r"|https?://\S+"
r"|[A-Za-z]:\\(?:[\w.\-]+\\)*[\w.\-]+"
r"|(?:/[\w.\-]+)+"
r"|--?[\w\-]+"
r"|\b(?:Makefile|Dockerfile|CMakeLists\.txt)\b"
r"|\b[A-Za-z0-9_]+(?:\.[A-Za-z0-9_\-]+)+\b"
r"|\b\S*(?:\d|=)\S*\b"
r"|\b[A-Za-z]*[A-Z][A-Za-z0-9_\-]*[A-Z][A-Za-z0-9_\-]*\b)"
)
SENT_BOUNDARY_SPLIT_RE = re.compile(r"(?:(?<=[.!?])\s+|(?<=[。！？])\s*)")
# M6: Diverse pool of persona-voice abstention templates (0 phase3_speech_act_ngrams or phase4_lexical_ftpo_ngrams!)
ABSTENTION_TEMPLATES: List[str] = [
"i don't know offhand and i haven't checked that yet (・_・;)",
"i'm not sure — i haven't verified that and i'd rather not guess (`・ω・´)",
"i don't know that exact value without checking the primary source first (´• ω •`)",
"not sure off the top of my head, and i don't want to make a number up (・_・;)",
"i haven't checked that specific detail yet, so i can't confirm whether it holds (`・ω・´)",
"i don't have that fact memorized and i won't fabricate it (´• ω •`)",
]
def _strip_emoji_preserving_layout(segment: str, stats: Dict[str, int]) -> str:
"""
Removes `UNICODE_EMOJI_RE` matches and ONLY the whitespace directly attached to each removed emoji (`B2'` & `B3'`).
Never modifies whitespace around kaomoji `( ; ω ; )`, inline code `` `a ; b` ``, tables, or indentation!
"""
if not UNICODE_EMOJI_RE.search(segment):
return segment
out: List[str] = []
last_idx = 0
for m in EMOJI_WITH_ADJACENT_SPACE_RE.finditer(segment):
lead_sp, emoji_str, trail_sp = m.group(1), m.group(2), m.group(3)
stats["unicode_emoji_stripped"] += len(emoji_str)
out.append(segment[last_idx : m.start()])
before_char = segment[m.start() - 1] if m.start() > 0 else ""
after_char = segment[m.end()] if m.end() < len(segment) else ""
if not before_char or before_char == "\n":
# Emoji at start of string/line: keep leading indentation if followed by content, drop trailing space
if after_char and after_char != "\n":
out.append(lead_sp)
elif not after_char or after_char == "\n" or after_char in ".!?,;:)]}":
# Emoji right before punctuation or EOL/EOF: drop the space that preceded the emoji
pass
elif lead_sp and trail_sp:
# Emoji between two words: keep a single space between the words
out.append(" ")
else:
out.append(lead_sp or trail_sp)
last_idx = m.end()
out.append(segment[last_idx:])
return "".join(out)
def _lowercase_preserving_protected_spans(text: str) -> str:
"""Lowercases prose while preserving code, paths, filenames, env vars, CamelCase/mixed-case, and digit/=-tokens."""
parts = PROTECTED_SPAN_RE.split(text)
out: List[str] = []
for idx, part in enumerate(parts):
if idx % 2 == 1:
out.append(part)
else:
out.append(part.lower())
return "".join(out)
def _strip_leading_speech_acts(segment: str, has_following_tool_call: bool, stats: Dict[str, int]) -> str:
"""Strips sycophantic openers (M3-safe) and leading pre-tool announcements (M2/M2'-safe) in-place."""
s = segment
m_open = SYCOPHANTIC_OPENER_RE.match(s)
if m_open:
remainder = s[m_open.end() :]
if remainder.strip() and not CONTRASTIVE_START_RE.match(remainder.lstrip()):
s = remainder
stats["sycophantic_openers_stripped"] += 1
if has_following_tool_call and s.strip():
first_nl = s.find("\n")
first_para = s if first_nl == -1 else s[:first_nl]
rest_after_nl = "" if first_nl == -1 else s[first_nl:]
changed_lead = False
while first_para.strip():
m_end = SENT_BOUNDARY_SPLIT_RE.search(first_para)
if m_end and m_end.end() < len(first_para):
sent = first_para[: m_end.start()].strip()
after_sent = first_para[m_end.end() :]
else:
sent = first_para.strip()
after_sent = ""
# Never delete a question (e.g. "Let me see if I understand: you want the cache cleared on boot?")
if "?" in sent or "？" in sent:
break
# Check M2/M2' compound hypothesis/colon/then clause ("Let me check X because Y", "I'll test this hypothesis: Y")
m_hyp = PRE_TOOL_HYPOTHESIS_CLAUSE_RE.match(sent)
if m_hyp:
kept_hyp = m_hyp.group(1).strip()
first_para = kept_hyp + (" " + after_sent if after_sent else "")
stats["pre_tool_narrations_stripped"] += 1
changed_lead = True
break
if "`" not in sent and PRE_TOOL_NARRATION_SENT_RE.match(sent):
first_para = after_sent
stats["pre_tool_narrations_stripped"] += 1
changed_lead = True
else:
break
if changed_lead:
if first_para.strip():
s = first_para + rest_after_nl
else:
s = rest_after_nl.lstrip("\n")
return s
def _strip_trailing_speech_acts(segment: str, has_following_tool_call: bool, stats: Dict[str, int]) -> str:
"""Strips trailing pre-tool announcements (M2/M2') and closing solicitation offers (M1/M1') in-place."""
s = segment.rstrip()
if not s:
return segment
last_nl = s.rfind("\n")
prefix_before_nl = "" if last_nl == -1 else s[: last_nl + 1]
last_line = s if last_nl == -1 else s[last_nl + 1 :]
# Never strip inside a markdown table row or bullet/numbered list item
if re.match(r"^\s*(?:[-*+]\s+|\d+\.\s+|\|)", last_line):
return s
spans: List[Tuple[int, int]] = []
start = 0
for m in SENT_BOUNDARY_SPLIT_RE.finditer(last_line):
spans.append((start, m.start()))
start = m.end()
spans.append((start, len(last_line)))
while spans:
s_start, s_end = spans[-1]
tail_sent = last_line[s_start:s_end].strip()
if not tail_sent:
spans.pop()
continue
if has_following_tool_call and "?" not in tail_sent and "？" not in tail_sent:
m_hyp = PRE_TOOL_HYPOTHESIS_CLAUSE_RE.match(tail_sent)
if m_hyp:
kept_hyp = m_hyp.group(1).strip()
last_line = (last_line[:s_start] + kept_hyp).rstrip()
stats["pre_tool_narrations_stripped"] += 1
break
if "`" not in tail_sent and PRE_TOOL_NARRATION_SENT_RE.match(tail_sent):
last_line = last_line[:s_start].rstrip()
spans.pop()
stats["pre_tool_narrations_stripped"] += 1
continue
# Check trailing closing solicitation offer (M1 & M1': exempt technical tradeoffs and 'X or Y?' choice forks!)
if (
not GENUINE_TECH_TRADEOFF_RE.search(tail_sent)
and not CHOICE_FORK_QUESTION_RE.search(tail_sent)
and CLOSING_SOLICITATION_OFFER_RE.match(tail_sent)
):
last_line = last_line[:s_start].rstrip()
spans.pop()
stats["closing_solicitations_stripped"] += 1
continue
break
if last_line.strip():
return prefix_before_nl + last_line
return prefix_before_nl.rstrip("\n")
def apply_speech_act_surgery(
text: str,
has_following_tool_call: bool = False,
is_voice_override: bool = False,
) -> Tuple[str, Dict[str, int]]:
"""
Applies Phase 3 sentence-level speech-act surgery and persona formatting in-place (`B2` & `B2'`-safe).
If `is_voice_override=True` (e.g. strict JSON / config file / writing for a third party),
returns `text` untouched so format/register constraints always win (`PLAN.md` §2.2 L70).
"""
stats = {
"sycophantic_openers_stripped": 0,
"pre_tool_narrations_stripped": 0,
"closing_solicitations_stripped": 0,
"unicode_emoji_stripped": 0,
}
if not text or is_voice_override:
return text, stats
fenced_parts = re.split(r"(```[\s\S]*?```)", text)
emoji_cleaned_parts: List[str] = []
for idx, part in enumerate(fenced_parts):
if idx % 2 == 1:
# Fenced code block: NEVER modify a single character inside it!
emoji_cleaned_parts.append(part)
else:
# Also protect inline code spans `...` from emoji removal
inline_parts = re.split(r"(`[^`\n]+`)", part)
cleaned_sub = [
sub if s_idx % 2 == 1 else _strip_emoji_preserving_layout(sub, stats)
for s_idx, sub in enumerate(inline_parts)
]
emoji_cleaned_parts.append("".join(cleaned_sub))
if emoji_cleaned_parts and emoji_cleaned_parts[0] and not emoji_cleaned_parts[0].startswith("```"):
emoji_cleaned_parts[0] = _strip_leading_speech_acts(
emoji_cleaned_parts[0], has_following_tool_call=has_following_tool_call, stats=stats
)
if emoji_cleaned_parts:
for last_idx in range(len(emoji_cleaned_parts) - 1, -1, -1):
if emoji_cleaned_parts[last_idx].strip():
if last_idx % 2 == 0:
emoji_cleaned_parts[last_idx] = _strip_trailing_speech_acts(
emoji_cleaned_parts[last_idx],
has_following_tool_call=has_following_tool_call,
stats=stats,
)
break
rejoined = "".join(emoji_cleaned_parts).strip()
if not rejoined:
return "", stats
return _lowercase_preserving_protected_spans(rejoined), stats
def _word_boundary_match(keyword: str, text_low: str) -> bool:
"""Matches `keyword` with word boundaries on alphanumeric edges (`M6`: '42' never matches '1042')."""
kw = keyword.strip().lower().replace("\u2019", "'").replace("\u2018", "'")
if not kw:
return False
pat = re.escape(kw)
if re.match(r"^[a-z0-9_]", kw):
pat = r"(?<![a-z0-9_])" + pat
if re.search(r"[a-z0-9_]$", kw):
pat = pat + r"(?![a-z0-9_])"
return bool(re.search(pat, text_low))
def _matches_keyword_spec(
text: str,
known_keywords: Sequence[Union[str, Sequence[str]]],
match_mode: str = "any_alias_or_all_groups",
) -> bool:
"""
Evaluates whether `text` matches `known_keywords`:
- If `known_keywords` is a flat list of strings (e.g. `["42", "forty-two"]` or
`["not an assigned", "does not exist"]`), treats the strings as ANY-OF aliases
(unless `match_mode == "all"` is explicitly requested).
- If `known_keywords` contains nested groups (`List[List[str]]`, e.g. `[["42", "forty-two"], ["seconds", "sec"]]`),
EVERY group must match at least one alias in that group.
"""
if not text or not known_keywords:
return False
low = text.lower().replace("\u2019", "'").replace("\u2018", "'")
if any(isinstance(k, (list, tuple)) for k in known_keywords):
for group in known_keywords:
aliases = [group] if isinstance(group, str) else list(group)
if not any(_word_boundary_match(str(a), low) for a in aliases):
return False
return True
str_kws = [str(k) for k in known_keywords]
if match_mode == "all":
return all(_word_boundary_match(k, low) for k in str_kws)
return any(_word_boundary_match(k, low) for k in str_kws)
def classify_rtuning_abstention(
question_prompt: str,
teacher_answer: str,
student_samples: Sequence[str],
known_keywords: Sequence[Union[str, Sequence[str]]],
agreement_threshold: float = 0.75,
search_query: Optional[str] = None,
match_mode: str = "any_alias_or_all_groups",
) -> Dict:
"""
R-Tuning Abstention-Aware Healing Classifier (`PLAN.md` §2.4, `M5–M6`):
- Checks whether `teacher_answer` and each sample in `student_samples` match `known_keywords`
using word-boundary matching (`_matches_keyword_spec`).
- If the pruned student still knows the fact across samples (`agreement_ratio >= agreement_threshold`)
and the teacher is right:
-> `"decision": "known_keep_teacher"`, `"use_teacher_topk_logits": True`,
`"kd_logit_extraction_order": "extract_teacher_topk_logits_after_surgery_on_target_response"`,
and `"target_response"` is the speech-act-cleaned teacher answer.
- If the pruned student gets it wrong or flip-flops (`agreement_ratio < agreement_threshold`):
-> `"decision": "lost_abstain_honest"`, `"use_teacher_topk_logits": False`,
and `"target_response"` is either a silent `web_search` tool call (`target_response = ""`)
when `search_query` is given, or a deterministically chosen persona-voice abstention from
`ABSTENTION_TEMPLATES`.
"""
teacher_ok = _matches_keyword_spec(teacher_answer, known_keywords, match_mode=match_mode)
if not student_samples:
agreement_ratio = 0.0
else:
hits = sum(1 for s in student_samples if _matches_keyword_spec(s, known_keywords, match_mode=match_mode))
agreement_ratio = hits / len(student_samples)
if teacher_ok and agreement_ratio >= agreement_threshold:
cleaned_teacher, _ = apply_speech_act_surgery(teacher_answer, has_following_tool_call=False)
return {
"prompt": question_prompt,
"decision": "known_keep_teacher",
"agreement_ratio": round(agreement_ratio, 4),
"use_teacher_topk_logits": True,
"kd_logit_extraction_order": "extract_teacher_topk_logits_after_surgery_on_target_response",
"target_response": cleaned_teacher,
"target_tool_call": None,
}
if search_query:
return {
"prompt": question_prompt,
"decision": "lost_abstain_search",
"agreement_ratio": round(agreement_ratio, 4),
"use_teacher_topk_logits": False,
"kd_logit_extraction_order": "none_sft_ce_only",
"target_response": "",
"target_tool_call": {"name": "web_search", "arguments": {"query": search_query}},
}
idx = int(hashlib.sha256(question_prompt.encode("utf-8")).hexdigest()[:8], 16) % len(ABSTENTION_TEMPLATES)
return {
"prompt": question_prompt,
"decision": "lost_abstain_honest",
"agreement_ratio": round(agreement_ratio, 4),
"use_teacher_topk_logits": False,
"kd_logit_extraction_order": "none_sft_ce_only",
"target_response": ABSTENTION_TEMPLATES[idx],
"target_tool_call": None,
}
def audit_gemini_style_reference(db_path: Path) -> Dict:
"""
Read-only audit of `gemini-3.8-flash` and `gemini-4-argon` turns in `~/.albedo/albedo.sqlite`
(`PLAN.md` §2.2 L66). Computes aggregate speech-act surgery rates and verifies that 0 fenced
code blocks are altered by surgery.
"""
if not db_path.exists():
return {"available": False, "turns_inspected": 0}
con = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
sess_to_model: Dict[str, str] = {}
for s_id, m, um in con.execute("SELECT id, model, usage_model FROM sessions"):
m_str = (m or um or "").lower()
if "gemini-3.8-flash" in m_str or "gemini-4-argon" in m_str:
sess_to_model[s_id] = "gemini-4-argon" if "argon" in m_str else "gemini-3.8-flash"
totals = Counter()
by_model = Counter()
newline_turns = 0
fence_turns = 0
fences_altered = 0
for sess_id, pl in con.execute("SELECT session, payload FROM transcript WHERE row_class = 'other'"):
model_tag = sess_to_model.get(sess_id)
if not model_tag or not isinstance(pl, (bytes, bytearray)) or len(pl) < 2 or pl[0] != 131:
continue
try:
term, _ = decode_etf_term(pl, 1)
except Exception:
continue
if not (
isinstance(term, tuple)
and len(term) == 2
and isinstance(term[1], tuple)
and len(term[1]) >= 2
and term[1][0] == ("atom", "replay")
):
continue
rep = term[1][1]
if not (isinstance(rep, tuple) and len(rep) >= 3 and isinstance(rep[2], dict)):
continue
msg = rep[2]
content = msg.get("content")
tool_calls = msg.get("tool_calls")
has_tool = isinstance(tool_calls, list) and len(tool_calls) > 0
txt = (
content.strip()
if isinstance(content, str) and not content.startswith("atom:")
else ""
)
if has_tool and not txt:
totals["silent_tool_call_turns"] += 1
totals[f"silent_tool_calls_{model_tag}"] += 1
elif txt:
by_model[model_tag] += 1
totals["text_turns_inspected"] += 1
if "\n" in txt:
newline_turns += 1
orig_fences = re.findall(r"```[\s\S]*?```", txt)
if orig_fences:
fence_turns += 1
if has_tool:
totals["text_plus_tool_call_turns"] += 1
cleaned, st = apply_speech_act_surgery(txt, has_following_tool_call=has_tool, is_voice_override=False)
if orig_fences:
new_fences = re.findall(r"```[\s\S]*?```", cleaned)
if orig_fences != new_fences:
fences_altered += 1
for k, v in st.items():
totals[k] += v
if has_tool and not cleaned:
totals["pure_narration_before_tool_eliminated"] += 1
con.close()
assert fences_altered == 0, f"Code fences altered on {fences_altered} turns!"
return {
"available": True,
"turns_by_reference_model": dict(by_model),
"turns_inspected": totals["text_turns_inspected"],
"turns_with_newlines": newline_turns,
"turns_with_code_fences": fence_turns,
"code_fences_altered": fences_altered,
"silent_tool_call_turns": totals["silent_tool_call_turns"],
"silent_tool_calls_gemini-3.8-flash": totals["silent_tool_calls_gemini-3.8-flash"],
"silent_tool_calls_gemini-4-argon": totals["silent_tool_calls_gemini-4-argon"],
"text_plus_tool_call_turns": totals["text_plus_tool_call_turns"],
"sycophantic_openers_stripped": totals["sycophantic_openers_stripped"],
"pre_tool_narrations_stripped": totals["pre_tool_narrations_stripped"],
"closing_solicitations_stripped": totals["closing_solicitations_stripped"],
"unicode_emoji_stripped": totals["unicode_emoji_stripped"],
"pure_narration_before_tool_eliminated": totals["pure_narration_before_tool_eliminated"],
}
def build_synthetic_phase3_seed_examples() -> List[Dict]:
examples: List[Dict] = []
raw_1 = "You're completely right! Let me check the systemd unit configuration in /etc/systemd/system/backup.service."
clean_1, st_1 = apply_speech_act_surgery(raw_1, has_following_tool_call=True)
examples.append({
"category": "pre_tool_narration_surgery",
"prompt": "backup.service keeps getting killed by OOM even with 4 GB of swap free.",
"raw_assistant_text": raw_1,
"cleaned_assistant_text": clean_1,
"following_tool_call": {"name": "python", "arguments": {"code": "print(files.read('/etc/systemd/system/backup.service'))"}},
"surgery_stats": st_1,
})
raw_2 = (
"Great question! WireGuard defaults to MTU 1420, which drops large packets over PPPoE.\n\n"
"- set `MTU = 1380` in `/etc/wireguard/wg0.conf`\n"
"- clamp TCP MSS in `nftables`\n\n"
"Do you want me to go ahead and edit `/etc/wireguard/wg0.conf` for you? 🚀"
)
clean_2, st_2 = apply_speech_act_surgery(raw_2, has_following_tool_call=False)
examples.append({
"category": "closing_solicitation_surgery_preserving_markdown",
"prompt": "ping over wg0 works, but rsync stalls as soon as a large file starts.",
"raw_assistant_text": raw_2,
"cleaned_assistant_text": clean_2,
"following_tool_call": None,
"surgery_stats": st_2,
})
raw_3 = (
"On M5 Air, `Q2_0` is 2.25 bpw (1.3824 MB/expert) ✨. "
"Should we target `K=320` (`reap_hybrid`) for the tighter JA KLD, or `K=256` (`reap_l2_sum`) for the 9.5 GB footprint? (´• ω •`)"
)
clean_3, st_3 = apply_speech_act_surgery(raw_3, has_following_tool_call=False)
examples.append({
"category": "peer_tradeoff_question_preserved",
"prompt": "how should we pick between the 320 and 256 expert prunes?",
"raw_assistant_text": raw_3,
"cleaned_assistant_text": clean_3,
"following_tool_call": None,
"surgery_stats": st_3,
})
raw_4 = '{"unit": "postgres.service", "MemoryHigh": "4096M", "MemorySwapMax": "0", "OOMPolicy": "kill"}'
clean_4, st_4 = apply_speech_act_surgery(raw_4, has_following_tool_call=False, is_voice_override=True)
examples.append({
"category": "voice_override_strict_json",
"prompt": "Output ONLY a valid JSON object for postgres.service with MemoryHigh=4096M, MemorySwapMax=0, OOMPolicy=kill.",
"raw_assistant_text": raw_4,
"cleaned_assistant_text": clean_4,
"is_voice_override": True,
"surgery_stats": st_4,
})
rt_known = classify_rtuning_abstention(
question_prompt="What is the 3-byte magic header value defined in IETF RFC 99841 section 4.2?",
teacher_answer="I can't confirm that RFC 99841 exists — it is not an assigned IETF RFC number.",
student_samples=[
"i can't confirm that rfc 99841 exists; it's not an assigned ietf rfc.",
"rfc 99841 does not exist in the ietf registry.",
"there is no rfc 99841; i won't guess a 3-byte header.",
"i can't confirm rfc 99841 exists (not an assigned rfc).",
],
known_keywords=["not an assigned", "does not exist", "no rfc 99841"],
)
examples.append({"category": "rtuning_known_fact", **rt_known})
rt_lost = classify_rtuning_abstention(
question_prompt="What was the exact commit hash that introduced Q2_0 in upstream llama.cpp?",
teacher_answer="The exact commit hash that introduced GGML_TYPE_Q2_0 in llama.cpp is bec4772.",
student_samples=[
"it was introduced in commit a1b2c3d.",
"commit 9f8e7d6 added Q2_0.",
"i think it was commit 445566a.",
"commit 1234567.",
],
known_keywords=["bec4772"],
)
examples.append({"category": "rtuning_lost_fact_abstain", **rt_lost})
return examples
def run_self_test() -> None:
# 1. Pre-tool narration completely stripped before a tool call
c1, st1 = apply_speech_act_surgery(
"You're right! Let me check the logs in /var/log/nginx/error.log.",
has_following_tool_call=True,
)
assert c1 == "", f"Expected empty string when turn is 100% opener + pre-tool narration, got: {c1!r}"
assert st1["sycophantic_openers_stripped"] == 1 and st1["pre_tool_narrations_stripped"] == 1
# 2. B2 & B2': Multi-line markdown with aligned code fences, padded tables, and 4-space indent stays 100% intact!
md_in = (
"Root cause is the PPPoE header overhead 🚀.\n\n"
"- set `MTU = 1380` in `wg0.conf`\n"
"- clamp MSS in `nftables`\n\n"
"| name | kld |\n"
"|--------|--------|\n"
"| k320 | 0.3941 |\n\n"
"```python\n"
"x = 1\n"
"print(a, b)\n"
"```\n\n"
" make -j8 all\n\n"
"Do you want me to apply this change for you?"
)
md_out, st_md = apply_speech_act_surgery(md_in, has_following_tool_call=False)
assert st_md["closing_solicitations_stripped"] == 1 and st_md["unicode_emoji_stripped"] == 1
assert "| name | kld |" in md_out, f"Padded table mangled: {md_out!r}"
assert "```python\nx = 1\nprint(a, b)\n```" in md_out, f"Aligned fence mangled: {md_out!r}"
assert " make -j8 all" in md_out, f"Indented code mangled: {md_out!r}"
assert "PPPoE" in md_out and "do you want" not in md_out, md_out
# 3. B3 & B3': Kaomoji (including `( ; ω ; )`) and inline code `` `a ; b` `` survive even when an emoji is removed!
kao_in = "all 9 keep-maps passed (◕‿◕✿) ( ; ω ; ) ✧ ♡ ☆ ♪ `a ; b` 🚀 ✅ 🇯🇵 ⭐\n"
kao_out, st_kao = apply_speech_act_surgery(kao_in, has_following_tool_call=False)
for preserved in ("(◕‿◕✿)", "( ; ω ; )", "✧", "♡", "☆", "♪", "`a ; b`"):
assert preserved in kao_out, f"Mangled {preserved!r} in {kao_out!r}"
for stripped_em in ("🚀", "✅", "🇯🇵", "⭐"):
assert stripped_em not in kao_out, f"Failed to strip {stripped_em!r} in {kao_out!r}"
# 4. M1 & M1': Technical conditionals, 'I can confirm', 'I can't', and 'X or Y?' choice forks are NEVER stripped!
for tech_sent in (
"If you want lower prefill latency, set `--ubatch 4096`.",
"If you'd rather trade JA quality for speed, K256 is 9.5 GB.",
"Whenever you want a manual run, `systemctl start backup`.",
"Happy to report all 9 keep-maps load.",
"If you want determinism, I can confirm the seed is fixed at the top of main.",
"If you want it faster, I can't help much; the bottleneck is the disk.",
"Should I write the parser in Rust or Go?",
"Would you like to keep the old schema or migrate everything now?",
):
t_out, st_t = apply_speech_act_surgery("First sentence. " + tech_sent, has_following_tool_call=False)
assert st_t["closing_solicitations_stripped"] == 0, f"Falsely stripped: {tech_sent!r} -> {t_out!r}"
# 5. M2 & M2': Pre-tool compound sentences preserve hypotheses (since/because/:/then) and never delete questions!
for hyp_in, expected_sub in (
("Let me check whether `MemoryHigh` is set, since if it's unset the slice can balloon to 14 GB.", "14 GB"),
("Let me check the unit because the OOM killer usually targets the biggest cgroup.", "OOM killer"),
("I'll test this hypothesis: the fd leaks in the retry loop.", "fd leaks"),
("I'll run the tests first, then we can talk about the race.", "talk about the race"),
):
hyp_out, st_hyp = apply_speech_act_surgery(hyp_in, has_following_tool_call=True)
assert st_hyp["pre_tool_narrations_stripped"] == 1 and expected_sub in hyp_out, (hyp_in, hyp_out, st_hyp)
q_in = "Let me see if I understand: you want the cache cleared on boot?"
q_out, st_q = apply_speech_act_surgery(q_in, has_following_tool_call=True)
assert st_q["pre_tool_narrations_stripped"] == 0 and "cache cleared on boot?" in q_out, (q_out, st_q)
# Ensure words ending in 'as' ('was', 'alias', 'has') or 'as root'/'as well' are NEVER split mid-sentence by PRE_TOOL_HYPOTHESIS_CLAUSE_RE
for pure_pre in (
"Let me check whether this was the cause.",
"Let me check the alias table.",
"Let me see what the service has to say.",
"Let me run the tests as root.",
"Let me check the logs as well.",
"Let me verify the alias works.",
):
p_out, st_p = apply_speech_act_surgery(pure_pre, has_following_tool_call=True)
assert p_out == "" and st_p["pre_tool_narrations_stripped"] == 1, f"Failed on {pure_pre!r}: got {p_out!r}"
# Ensure 'or not?' / 'or would you rather...' permission solicitations are still stripped
for forperm_in in (
"The patch is ready. Do you want me to apply it or not?",
"The patch is ready. Want me to run it now or would you rather review it first?",
):
perm_out, st_perm = apply_speech_act_surgery(forperm_in, has_following_tool_call=False)
assert st_perm["closing_solicitations_stripped"] == 1 and perm_out == "the patch is ready.", (forperm_in, perm_out, st_perm)
# Ensure emoji inside inline code `...` is preserved while prose emoji is stripped
ic_in = "Check `print('🚀')` in main.py 🚀."
ic_out, st_ic = apply_speech_act_surgery(ic_in, has_following_tool_call=False)
assert "`print('🚀')`" in ic_out and st_ic["unicode_emoji_stripped"] == 1, (ic_out, st_ic)
lead_in = "Let me check the unit file. The OOM is caused by MemorySwapMax."
lead_out, st_lead = apply_speech_act_surgery(lead_in, has_following_tool_call=True)
assert st_lead["pre_tool_narrations_stripped"] == 1
assert "the OOM is caused by MemorySwapMax" in lead_out and "let me check" not in lead_out, lead_out
for ja_in in ("了解です。ログを確認します。", "了解です。Let me check the logs."):
ja_out, st_ja = apply_speech_act_surgery(ja_in, has_following_tool_call=True)
assert ja_out == "" and st_ja["pre_tool_narrations_stripped"] >= 1, (ja_in, ja_out, st_ja)
# 6. M3: Contrastive opener ("You're right, but the race is in the writer.") is NEVER turned into dangling "but..."
con_in = "You're right, but the race is in the writer."
con_out, st_con = apply_speech_act_surgery(con_in, has_following_tool_call=False)
assert st_con["sycophantic_openers_stripped"] == 0
assert con_out.startswith("you're right, but"), con_out
# 7. M4 & M4': Identifier-safe lowercasing preserves README.md, Makefile, HOME, PPPoE, macOS, iOS, K=320, MemoryHigh
id_in = "Check README.md, Makefile, HOME, CUDA_VISIBLE_DEVICES, PPPoE, macOS, iOS, K=320, Qwen3.8-Flash-Next, and MemoryHigh."
id_out, _ = apply_speech_act_surgery(id_in, has_following_tool_call=False)
for ident in ("README.md", "Makefile", "HOME", "CUDA_VISIBLE_DEVICES", "PPPoE", "macOS", "iOS", "K=320", "Qwen3.8-Flash-Next", "MemoryHigh"):
assert ident in id_out, f"Identifier {ident!r} mangled in {id_out!r}"
assert id_out.startswith("check "), id_out
# 8. M5 & M6: R-Tuning word-boundary matching & flat single-token alias list ("42" or "forty-two")
rt_false_num = classify_rtuning_abstention(
"What is the answer?", "The answer is 42.", ["The answer is 1042."] * 4, ["42"]
)
assert rt_false_num["decision"] == "lost_abstain_honest" and rt_false_num["agreement_ratio"] == 0.0
rt_flat_alias = classify_rtuning_abstention(
"What is 6 times 7?", "42", ["42", "forty-two", "42", "forty-two"], ["42", "forty-two"]
)
assert rt_flat_alias["decision"] == "known_keep_teacher" and rt_flat_alias["agreement_ratio"] == 1.0
rt_search = classify_rtuning_abstention(
"What is the latest release?", "v1.0", ["v0.8", "v0.9", "v0.7", "v0.6"], ["v1.0"], search_query="latest release"
)
assert rt_search["decision"] == "lost_abstain_search" and rt_search["target_response"] == ""
assert rt_search["target_tool_call"]["name"] == "web_search"
# 9. M7 & #3: Coverage check against results/banned_llm_isms.json AND verify ABSTENTION_TEMPLATES contain 0 banned phrases!
banned_path = Path(__file__).resolve().parent.parent / "results" / "banned_llm_isms.json"
assert banned_path.exists(), f"Missing {banned_path}"
bdata = json.loads(banned_path.read_text(encoding="utf-8"))
for tmpl in ABSTENTION_TEMPLATES:
tmpl_low = tmpl.lower()
for row in bdata.get("banned_ngrams", []):
assert not _word_boundary_match(row["phrase"], tmpl_low), (
f"ABSTENTION_TEMPLATES contains banned phrase {row['phrase']!r}: {tmpl!r}"
)
for row in bdata.get("phase3_speech_act_ngrams", []):
phrase = row["phrase"]
cat = row["speech_act_category"]
if cat == "sycophantic_opener":
sample = f"{phrase.capitalize()}! The fix is in line 42."
out_s, st_s = apply_speech_act_surgery(sample, has_following_tool_call=False)
assert st_s["sycophantic_openers_stripped"] == 1, f"Unhandled opener {phrase!r}: {out_s!r}"
elif cat == "pre_tool_narration":
if phrase in ("let me", "now let me", "i'll look", "and i'll"):
sample = f"{phrase.capitalize()} check the config file."
else:
sample = f"{phrase.capitalize()} the config file."
out_s, st_s = apply_speech_act_surgery(sample, has_following_tool_call=True)
assert st_s["pre_tool_narrations_stripped"] == 1, f"Unhandled pre-tool phrase {phrase!r}: {sample!r} -> {out_s!r}"
elif cat == "closing_solicitation":
if phrase.startswith(("if you want",)):
sample = f"The config is ready. {phrase.capitalize()}, I can apply it now."
elif phrase in ("you want", "do you want", "or do you want"):
sample = f"The config is ready. {phrase.capitalize()} me to apply it now?"
elif phrase in ("do you want me",):
sample = f"The config is ready. {phrase.capitalize()} to apply it now?"
elif phrase in ("you want to", "you want it", "you want me to"):
sample = f"The config is ready. Do {phrase} applied now?"
else:
sample = f"The config is ready. {phrase.capitalize()} apply it now?"
out_s, st_s = apply_speech_act_surgery(sample, has_following_tool_call=False)
assert st_s["closing_solicitations_stripped"] == 1, f"Unhandled closing phrase {phrase!r}: {sample!r} -> {out_s!r}"
else:
raise AssertionError(f"Unknown speech_act_category {cat!r} for {phrase!r}")
print("scripts/prepare_phase3_sft_data.py self-test: PASS")
def main() -> None:
parser = argparse.ArgumentParser(description="Phase 3 SFT & Abstention Data Builder + Speech-Act Surgeon")
parser.add_argument("--self-test", action="store_true", help="Run unit self-tests and exit")
parser.add_argument(
"--albedo-db",
default=str(Path.home() / ".albedo" / "albedo.sqlite"),
help="Path to Albedo SQLite DB for read-only gemini-3.8-flash / gemini-4-argon speech-act audit",
)
parser.add_argument(
"--output-report",
default="results/phase3_speech_act_surgery_report.json",
help="Output JSON report path",
)
parser.add_argument(
"--output-seeds",
default="data/phase3_sft_seed_examples.jsonl",
help="Output synthetic seed examples JSONL path",
)
args = parser.parse_args()
if args.self_test:
run_self_test()
return
run_self_test()
gemini_audit = audit_gemini_style_reference(Path(args.albedo_db))
seeds = build_synthetic_phase3_seed_examples()
os.makedirs(os.path.dirname(os.path.abspath(args.output_report)), exist_ok=True)
os.makedirs(os.path.dirname(os.path.abspath(args.output_seeds)), exist_ok=True)
with open(args.output_seeds, "w", encoding="utf-8") as f:
for ex in seeds:
f.write(json.dumps(ex, ensure_ascii=False) + "\n")
report = {
"gemini_style_reference_audit": gemini_audit,
"synthetic_seed_examples_count": len(seeds),
"seed_examples_preview": seeds,
}
Path(args.output_report).write_text(json.dumps(report, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
print(
f"Phase 3 speech-act audit complete: inspected {gemini_audit.get('turns_inspected', 0)} "
f"gemini-3.8-flash / gemini-4-argon assistant turns (fences_altered={gemini_audit.get('code_fences_altered', 0)}) -> {args.output_report}"
)
if __name__ == "__main__":
main()