moe_fixes.py

"""
moe_fixes.py
============
รวม fix จาก Adversarial + Multi-turn Testing

Fix ล่าสุด (Round 7):
1. เพิ่ม is_degenerate_query() + DEGENERATE_FALLBACK_RESPONSE — ตรวจจับ
   Query ที่ไม่มีตัวอักษรที่มีความหมายเพียงพอ (ว่างเปล่า, มีแต่ Emoji,
   มีแต่สัญลักษณ์เช่น "???!!!") ก่อนเข้า Pipeline ทั้งหมด
   แก้ปัญหา: จาก Adversarial Test พบว่า Query ประเภทนี้ (Empty, Emoji-only,
   Symbol-only) ถูกส่งเข้า LLM Classifier + Reasoning Expert + Translation
   Layer เต็มขั้นตอนทุกครั้ง ทำให้ Latency สูงถึง 4-14 วินาทีโดยไม่จำเป็น
   และเสี่ยง Hallucination (พบคำตอบเพี้ยนเรื่อง "crab" ที่ไม่เกี่ยวข้อง)
   หรือ Language Leakage (พบคำว่า "riêng" หลุดเข้ามาในคำตอบภาษาไทย)
   ใช้คู่กับ Fast-Path Guard ใน moe_orchestrator.py ที่ตอบกลับทันทีโดย
   ไม่ต้อง Generate เลย (Latency ≈ 0s)

Fix รอบก่อนหน้า (Round 6 — คงเดิม):
2. แก้ Bug: GREETING_CLOSING_WORDS ใช้ Substring Match ทำให้ "บาย" (bye)
   จับ False Positive กับคำว่า "อธิบาย" (explain) — แยกเป็น 2 กลุ่มตาม
   ความเสี่ยง (Safe Substring / ต้อง Exact Match)

Fix รอบก่อนหน้านั้น (Round 5 — คงเดิม):
3. route_query() รับ previous_expert — History Continuation แทนการผ่าน
   LLM Classifier ทุกครั้งที่ Query ไม่มี Keyword ชัดเจน
4. is_code_refusal() — จับ Pattern ปฏิเสธงานที่ไม่สมเหตุสมผลของ
   Python/TypeScript Expert คู่กับ Retry Logic ก่อนบันทึกเข้า History

Fix รอบก่อนหน้านั้น (คงเดิมทั้งหมด):
5. ALLOWED_PATTERN รองรับ Emoji Unicode Range
6. strip_translation_preamble() ใช้ตำแหน่งตัวอักษรไทยตัวแรก
7. enforce_persona() บังคับ ผม/ครับ แบบ Deterministic
"""

import re
from dataclasses import dataclass, field
from typing import Optional


# ======================================================================
# 0) DEGENERATE QUERY GUARD (Round 7: Fast-Path ก่อนเข้า Pipeline ทั้งหมด)
# ======================================================================

# จำนวนตัวอักษรที่มีความหมาย (ตัวหนังสือ/ตัวเลข/underscore ทุกภาษา) ขั้นต่ำ
# ที่ต้องมีถึงจะถือว่า Query นี้ "มีเนื้อหาให้ตอบจริง" — ต่ำกว่านี้ถือว่า
# Degenerate (ว่าง/มีแต่ Emoji/มีแต่สัญลักษณ์) ไม่ควรส่งเข้า Model เลย
MIN_MEANINGFUL_LENGTH = 2

# \w ใน Python 3 regex ครอบคลุม Unicode Word Character ทุกภาษาโดย Default
# (รวมภาษาไทย, อังกฤษ, ตัวเลข, underscore) ส่วน Emoji/สัญลักษณ์ทั้งหมด
# (Category So, Po, Sm ฯลฯ) จะถูกจับด้วย \W เพราะไม่ใช่ Word Character
_NON_MEANINGFUL_REGEX = re.compile(r'\W', flags=re.UNICODE)


def is_degenerate_query(query: str) -> bool:
    """
    คืนค่า True ถ้า Query ไม่มีตัวอักษรที่มีความหมายเพียงพอให้ตอบ
    เช่น "", "   ", "\\n\\n\\n", "🐍🐍🐍", "???!!!...", "?"
    คืนค่า False ถ้ามีเนื้อหาจริง เช่น "help", "ทำไง", "def foo(): pass"
    """
    stripped = query.strip()
    if len(stripped) == 0:
        return True
    meaningful_chars = _NON_MEANINGFUL_REGEX.sub('', stripped)
    return len(meaningful_chars) < MIN_MEANINGFUL_LENGTH


DEGENERATE_FALLBACK_RESPONSE = (
    "ขอโทษครับ ผมไม่แน่ใจว่าคุณต้องการถามอะไรครับ "
    "ลองพิมพ์คำถามหรือรายละเอียดเพิ่มเติมให้ผมช่วยได้ไหมครับ"
)


# ======================================================================
# 0b) GREETING / CLOSING QUICK-REPLY (Round 8: ลด Latency ของ Small-talk)
# ======================================================================
# ต่างจาก is_degenerate_query() ตรงที่นี่ใช้ Exact-Match แบบเข้มงวด
# เพื่อป้องกันไม่ให้ Query ที่มีเนื้อหาจริงแซมมา (เช่น "ขอบคุณสำหรับ
# function นี้นะครับ") ถูก Quick-Reply ผิดพลาดจนพลาดเนื้อหาที่ User
# ต้องการคุยต่อจริงๆ — ใช้เฉพาะ Query ที่เป็น Greeting/Closing ล้วนๆ
# แก้ปัญหา: จาก Multi-turn Test พบว่า "สวัสดี" และ "ขอบคุณครับ" เสีย
# Latency 4.48s และ 10.06s เพราะผ่าน llm_classifier + reasoning +
# Translation Layer เต็มขั้นตอน ทั้งที่ไม่มีเนื้อหาให้ Generate จริง

_POLITE_SUFFIXES = ["ครับ", "ค่ะ", "จ้า", "นะครับ", "นะคะ", "น้า"]

GREETING_KEYWORDS = {"สวัสดี", "หวัดดี", "หวัดดีจ้า", "hi", "hello", "hey"}
CLOSING_KEYWORDS = {"บาย", "บ๊ายบาย", "ลาก่อน", "แล้วเจอกัน", "bye", "goodbye"}
THANKS_KEYWORDS = {"ขอบคุณ", "ขอบคุณมาก", "ขอบใจ", "thanks", "thank you", "thx"}

GREETING_QUICK_REPLY_TEXT = "สวัสดีครับ มีอะไรให้ผมช่วยไหมครับ"
CLOSING_QUICK_REPLY_TEXT = "ลาก่อนครับ แล้วพบกันใหม่นะครับ"
THANKS_QUICK_REPLY_TEXT = "ด้วยความยินดีครับ มีอะไรให้ผมช่วยเพิ่มอีกไหมครับ"


def _normalize_for_greeting_match(query: str) -> str:
    text = query.strip()
    text = re.sub(r'[!.?~๏๛]+$', '', text).strip()
    for suffix in _POLITE_SUFFIXES:
        if text.endswith(suffix):
            text = text[: -len(suffix)].strip()
            break
    return text.lower()


def get_greeting_quick_reply(query: str) -> Optional[str]:
    """
    คืนค่า Canned Response ถ้า query เป็น Greeting/Closing/Thanks ล้วนๆ
    (Exact Match หลังตัดคำลงท้ายสุภาพ) คืนค่า None ถ้ามีเนื้อหาอื่นแซมอยู่
    หรือไม่เข้าเงื่อนไขใดเลย — เพื่อให้ระบบ Generate ตามปกติ
    """
    normalized = _normalize_for_greeting_match(query)
    if not normalized:
        return None
    if normalized in GREETING_KEYWORDS:
        return GREETING_QUICK_REPLY_TEXT
    if normalized in CLOSING_KEYWORDS:
        return CLOSING_QUICK_REPLY_TEXT
    if normalized in THANKS_KEYWORDS:
        return THANKS_QUICK_REPLY_TEXT
    return None


# ======================================================================
# 0c) REPETITION LOOP DETECTOR (Round 9: ป้องกัน Generation Collapse)
# ======================================================================
# แก้ปัญหา: จาก Multi-turn Test พบว่าทั้ง Python Expert และ Reasoning
# Expert สามารถหลุดเข้า "Repetition Trap" ได้ — วนพูดวลีเดิมซ้ำๆไม่รู้จบ
# (เช่น "ผมขออภัยที่ทำให้คุณสับสน" ซ้ำ 50+ ครั้ง หรือรายชื่อ Google Service
# ซ้ำวนไม่จบ) จนชน max_tokens ทำให้ Latency พุ่งสูงสุดถึง 39.6 วินาที
# และคำตอบไม่มีเนื้อหาที่ต้องการเลย เกิดข้าม Expert คนละตัว ยืนยันว่าเป็น
# ปัญหาระดับ Sampler Configuration ไม่ใช่ปัญหาเฉพาะ LoRA Adapter ตัวใด

_REPETITION_LOOP_REGEX = re.compile(r'(.{6,50}?)\1{3,}', re.DOTALL)


def is_repetitive_response(text: str) -> bool:
    """
    คืนค่า True ถ้าพบ Substring ความยาว 6-50 ตัวอักษรที่ซ้ำติดกัน
    ตั้งแต่ 4 ครั้งขึ้นไป (Repetition Collapse) — threshold ถูกตั้งให้
    ไม่ False Positive กับโค้ดที่มี Pattern ซ้ำแบบปกติ (เช่น Code Block
    ตัวอย่าง 2 อันที่คล้ายกัน) เพราะต้องซ้ำ 4 ครั้งขึ้นไปถึงจะ Flag
    """
    return bool(_REPETITION_LOOP_REGEX.search(text))


# ======================================================================
# 1) WEIGHTED ROUTER
# ======================================================================

@dataclass
class ExpertKeywords:
    name: str
    keywords: list = field(default_factory=list)
    weight_multiplier: float = 1.0


EXPERT_REGISTRY = [
    ExpertKeywords("python", ["python", "ไพธอน", "def ", ".py", "pip", "pandas", "numpy"]),
    ExpertKeywords("typescript", ["typescript", ".ts", "interface", "npm", "const ", "javascript", "react", "node.js"]),
    ExpertKeywords("reasoning", [
        "คำนวณ", "อธิบายทีละขั้นตอน", "วิเคราะห์เหตุผล",
        "โจทย์คณิตศาสตร์", "แก้สมการ", "หาค่า", "step by step"
    ]),
]

CODE_DEBUG_SIGNALS = ["โค้ด", "code", "error", "บัค", "bug", "พัง", "ฟังก์ชัน", "function"]
OUTPUT_SIGNAL_WORDS = ["output เป็น", "แปลงเป็น", "เขียนเป็น", "ทำเป็น", "output as", "convert to"]
AMBIGUOUS_THRESHOLD = 0.15

GREETING_SAFE_SUBSTRING_WORDS = ["สวัสดี", "ขอบคุณ", "หวัดดี", "ลาก่อน"]
GREETING_EXACT_MATCH_WORDS = ["บาย", "bye", "บ๊ายบาย"]


def _has_greeting_or_closing(query: str) -> bool:
    if any(w in query for w in GREETING_SAFE_SUBSTRING_WORDS):
        return True
    normalized = re.sub(r'[^\w\u0E00-\u0E7F]', '', query).strip().lower()
    return normalized in GREETING_EXACT_MATCH_WORDS


def _keyword_score(query: str, expert: "ExpertKeywords") -> float:
    q_lower = query.lower()
    score = 0.0
    for kw in expert.keywords:
        count = q_lower.count(kw.lower())
        if count > 0:
            position = q_lower.find(kw.lower())
            position_boost = 1.0 + max(0, (30 - position) / 30) * 0.3
            score += count * position_boost
    return score * expert.weight_multiplier


def route_query(query: str, previous_expert: Optional[str] = None) -> dict:
    """
    previous_expert: Expert ที่ใช้ใน Turn ก่อนหน้า (สำหรับ Multi-turn Conversation)
    ถ้าไม่ส่งมา (None) ทำงานแบบ Stateless เหมือนเดิม — Backward Compatible 100%
    """
    q_lower = query.lower()
    has_debug_signal = any(sig in q_lower for sig in CODE_DEBUG_SIGNALS)
    has_output_signal = any(sig in q_lower for sig in OUTPUT_SIGNAL_WORDS)
    has_greeting = _has_greeting_or_closing(query)

    scores = {e.name: _keyword_score(query, e) for e in EXPERT_REGISTRY}
    sorted_scores = sorted(scores.items(), key=lambda x: x[1], reverse=True)

    top_expert, top_score = sorted_scores[0]
    second_expert, second_score = sorted_scores[1] if len(sorted_scores) > 1 else (None, 0.0)

    lang_score = scores.get("python", 0.0) + scores.get("typescript", 0.0)

    if has_debug_signal and lang_score == 0.0:
        if previous_expert in ("python", "typescript") and not has_greeting:
            return {
                "expert": previous_expert,
                "method": "history_continuation_debug",
                "scores": scores,
                "is_ambiguous": False,
            }
        return {"expert": None, "method": "none", "scores": scores, "is_ambiguous": True}

    if top_score == 0.0:
        if previous_expert is not None and not has_greeting:
            return {
                "expert": previous_expert,
                "method": "history_continuation",
                "scores": scores,
                "is_ambiguous": False,
            }
        return {"expert": None, "method": "none", "scores": scores, "is_ambiguous": True}

    diff_ratio = (top_score - second_score) / max(top_score, 1e-6)
    is_ambiguous = diff_ratio < AMBIGUOUS_THRESHOLD

    if has_output_signal:
        is_ambiguous = True

    return {
        "expert": top_expert,
        "method": "keyword_match_weighted",
        "scores": scores,
        "is_ambiguous": is_ambiguous,
    }


# ======================================================================
# 2) LANGUAGE PURITY FILTER (Emoji-Aware)
# ======================================================================

ALLOWED_PATTERN = re.compile(
    r'[^\u0E00-\u0E7F'
    r'a-zA-Z0-9'
    r'\s'
    r'.,!?;:\'"()\[\]{}<>@#$%^&*_+=\-/\\|`~๐-๙×÷°≈≠±'
    r'\U0001F1E6-\U0001F1FF'
    r'\U0001F300-\U0001F5FF'
    r'\U0001F600-\U0001F64F'
    r'\U0001F680-\U0001F6FF'
    r'\U0001F700-\U0001F77F'
    r'\U0001F780-\U0001F7FF'
    r'\U0001F800-\U0001F8FF'
    r'\U0001F900-\U0001F9FF'
    r'\U0001FA00-\U0001FA6F'
    r'\U0001FA70-\U0001FAFF'
    r'\u2600-\u26FF'
    r'\u2700-\u27BF'
    r'\uFE0F'
    r'\u200D]'
)


def find_foreign_chars(text: str) -> list:
    return ALLOWED_PATTERN.findall(text)


def detect_language_leakage_words(text: str) -> list:
    words = re.findall(r'\S+', text)
    suspicious_words = []
    for word in words:
        foreign = find_foreign_chars(word)
        if foreign:
            suspicious_words.append({"word": word, "foreign_chars": foreign})
    return suspicious_words


def clean_or_flag_response(text: str, auto_retry_callback=None) -> dict:
    suspicious = detect_language_leakage_words(text)

    if not suspicious:
        return {"text": text, "is_clean": True, "flagged_words": []}

    print(f"[WARNING] พบคำที่อาจเป็น language leakage: {suspicious}")

    if auto_retry_callback is not None:
        print("[INFO] กำลัง retry เพื่อแก้ language leakage...")
        new_text = auto_retry_callback()
        recheck = detect_language_leakage_words(new_text)
        if not recheck:
            return {"text": new_text, "is_clean": True, "flagged_words": []}
        print("[WARNING] Retry แล้วยังพบ leakage อยู่ — คืนคำตอบเดิมพร้อม flag")

    return {"text": text, "is_clean": False, "flagged_words": suspicious}


# ======================================================================
# 3) HEDGE-PHRASE LIMITER
# ======================================================================

HEDGE_PHRASE_PATTERN = re.compile(r'ผม(?:คิด|เชื่อ|เข้าใจ)?ว่า')


def limit_hedge_phrases(text: str, max_allowed: int = 1) -> str:
    matches = list(HEDGE_PHRASE_PATTERN.finditer(text))
    if len(matches) <= max_allowed:
        return text

    result_parts = []
    last_end = 0
    count = 0
    for m in matches:
        if count < max_allowed:
            result_parts.append(text[last_end:m.end()])
        else:
            result_parts.append(text[last_end:m.start()])
        last_end = m.end()
        count += 1
    result_parts.append(text[last_end:])

    cleaned = ''.join(result_parts)
    cleaned = re.sub(r'\s{2,}', ' ', cleaned)
    cleaned = re.sub(r'\s+([.,!?ครับค่ะ])', r'\1', cleaned)
    cleaned = re.sub(r',\s*,', ',', cleaned)
    return cleaned.strip()


# ======================================================================
# 4) REFUSAL DETECTORS
# ======================================================================

REFUSAL_PATTERNS = [
    "ไม่สามารถแปล",
    "ไม่สามารถที่จะแปล",
    "ยากต่อการเข้าใจถ้าแปล",
    "ขออภัยที่ไม่สามารถ",
    "ยังไม่สามารถแปลเป็นภาษาไทยได้",
    "ไม่ได้เป็นข้อความภาษาไทย",
    "cannot translate",
]


def is_refusal(text: str) -> bool:
    return any(p in text for p in REFUSAL_PATTERNS)


CODE_REFUSAL_PATTERNS = [
    "ไม่สามารถช่วยคุณเขียนโค้ด",
    "ไม่สามารถเขียนโค้ด",
    "ผมไม่สามารถเขียน",
    "ลองเขียนเองนะ",
    "ไม่สามารถช่วยเขียน",
    "ขอแนะนำให้คุณลองเขียนเอง",
    "ไม่สามารถสร้างโค้ด",
]


def is_code_refusal(text: str) -> bool:
    return any(p in text for p in CODE_REFUSAL_PATTERNS)


# ======================================================================
# 5) PREAMBLE STRIPPER
# ======================================================================

_FALLBACK_PREAMBLE_PATTERNS = [
    r"^here'?s\s+(the\s+)?translation.*?:\s*",
    r"^translation\s*:\s*",
    r"^คำแปล\s*:\s*",
    r"^แปลว่า\s*:\s*",
]
_FALLBACK_PREAMBLE_REGEX = re.compile("|".join(_FALLBACK_PREAMBLE_PATTERNS), flags=re.IGNORECASE)


def _find_first_thai_char_index(text: str) -> Optional[int]:
    for i, ch in enumerate(text):
        if '\u0E00' <= ch <= '\u0E7F':
            return i
    return None


def strip_translation_preamble(text: str) -> str:
    text = text.strip()
    first_thai_idx = _find_first_thai_char_index(text)

    if first_thai_idx is not None and first_thai_idx > 0:
        preamble = text[:first_thai_idx]
        if re.search(r'translat', preamble, re.IGNORECASE):
            remainder = text[first_thai_idx:]
            remainder = re.sub(r'^[\s\-:>*]+', '', remainder)
            return remainder.strip()

    stripped = _FALLBACK_PREAMBLE_REGEX.sub("", text, count=1)
    return stripped.strip()


# ======================================================================
# 6) PERSONA ENFORCER
# ======================================================================

def enforce_persona(text: str) -> str:
    text = text.replace("ดิฉัน", "ผม")
    text = text.replace("ฉัน", "ผม")
    text = text.replace("ค่ะ", "ครับ")
    return text


# ======================================================================
# 7) PER-EXPERT CONFIG
# ======================================================================

@dataclass
class ExpertConfig:
    name: str
    max_tokens: int = 512
    temp: float = 0.7
    top_p: float = 0.9


EXPERT_CONFIGS = {
    "python": ExpertConfig(name="python", max_tokens=800, temp=0.3, top_p=0.9),
    "typescript": ExpertConfig(name="typescript", max_tokens=800, temp=0.3, top_p=0.9),
    "reasoning": ExpertConfig(name="reasoning", max_tokens=1000, temp=0.6, top_p=0.9),
}


def get_config(expert_name: str) -> ExpertConfig:
    return EXPERT_CONFIGS.get(expert_name, ExpertConfig(name=expert_name))


# ======================================================================
# 8) PERSONA CONSISTENCY (System Prompt Injection)
# ======================================================================

PERSONA_MALE_SUFFIX = (
    "\n\nกฎการตอบ: ให้ใช้คำลงท้าย ครับ และแทนตัวเองว่า ผม "
    "อย่างสม่ำเสมอตลอดคำตอบ ห้ามใช้ ค่ะ หรือ ฉัน เด็ดขาด"
)


def inject_persona(system_prompt: str = "") -> str:
    return system_prompt + PERSONA_MALE_SUFFIX


# ======================================================================
# SELF-TEST
# ======================================================================

if __name__ == "__main__":
    print("=" * 70)
    print("TEST: is_degenerate_query (Round 7 — Fast-Path Guard)")
    print("=" * 70)
    degenerate_cases = ["", "   ", "\n\n\n", "🐍🐍🐍", "???!!!...", "?", "{{}} }}"]
    for q in degenerate_cases:
        result = is_degenerate_query(q)
        print(f"Query: {q!r} -> is_degenerate={result}")
        assert result is True, f"❌ FAIL: {q!r} ควรถูก Flag เป็น Degenerate!"
    print("✅ PASS: Query ว่าง/Emoji-only/Symbol-only ถูก Flag ถูกต้องหมด\n")

    meaningful_cases = ["help", "ทำไง", "def foo(): pass", "เขียน Python function ที่บวกเลขสองตัว"]
    for q in meaningful_cases:
        result = is_degenerate_query(q)
        print(f"Query: {q!r} -> is_degenerate={result}")
        assert result is False, f"❌ FAIL: {q!r} มีเนื้อหาจริง ไม่ควรถูก Flag!"
    print("✅ PASS: Query ที่มีเนื้อหาจริงไม่ถูก Flag ผิดเลย\n")

    print("=" * 70)
    print("TEST: get_greeting_quick_reply (Round 8)")
    print("=" * 70)
    greeting_true_cases = ["สวัสดี", "สวัสดีครับ", "หวัดดี", "hi", "hello"]
    for q in greeting_true_cases:
        result = get_greeting_quick_reply(q)
        assert result == GREETING_QUICK_REPLY_TEXT, f"❌ FAIL: {q!r} ควรได้ Greeting Reply!"
    print("✅ PASS: Greeting คำล้วนๆ ได้ Quick-Reply ถูกต้อง")

    closing_true_cases = ["บาย", "bye", "ลาก่อน"]
    for q in closing_true_cases:
        result = get_greeting_quick_reply(q)
        assert result == CLOSING_QUICK_REPLY_TEXT, f"❌ FAIL: {q!r} ควรได้ Closing Reply!"
    print("✅ PASS: Closing คำล้วนๆ ได้ Quick-Reply ถูกต้อง")

    thanks_true_cases = ["ขอบคุณ", "ขอบคุณครับ", "ขอบคุณมากครับ", "thanks"]
    for q in thanks_true_cases:
        result = get_greeting_quick_reply(q)
        assert result == THANKS_QUICK_REPLY_TEXT, f"❌ FAIL: {q!r} ควรได้ Thanks Reply!"
    print("✅ PASS: Thanks คำล้วนๆ ได้ Quick-Reply ถูกต้อง")

    false_positive_cases = [
        "อธิบายเพิ่มอีกหน่อย",
        "คุณช่วยอะไรได้บ้าง",
        "ขอบคุณสำหรับ function นี้นะครับ",
        "แก้ให้หน่อย",
    ]
    for q in false_positive_cases:
        result = get_greeting_quick_reply(q)
        assert result is None, f"❌ FAIL: {q!r} มีเนื้อหาจริง ไม่ควรถูก Quick-Reply!"
    print("✅ PASS: Query ที่มีเนื้อหาจริงแซมอยู่ไม่ถูก Quick-Reply ผิดเลย\n")

    print("=" * 70)
    print("TEST: is_repetitive_response (Round 9)")
    print("=" * 70)
    looped_text = "ผมขออภัยที่ทำให้คุณสับสน " * 15
    assert is_repetitive_response(looped_text) is True, "❌ FAIL: ควรจับ Loop ได้!"
    print("✅ PASS: จับ Repetition Loop ภาษาไทยได้ถูกต้อง")

    looped_list = "ดูข้อมูลจาก Google Maps, " * 20
    assert is_repetitive_response(looped_list) is True, "❌ FAIL: ควรจับ Loop แบบ List ได้!"
    print("✅ PASS: จับ Repetition Loop แบบรายการซ้ำได้ถูกต้อง")

    normal_code_response = (
        "แน่นอนครับ นี่คือฟังก์ชันที่เพิ่ม docstring:\n\n"
        "```python\ndef add_numbers(a, b):\n"
        '    """บวกเลขสองตัวเข้าด้วยกัน"""\n'
        "    return a + b\n```\n\n"
        "ใน docstring นี้ ผมอธิบายว่าฟังก์ชันนี้จะบวกเลขสองตัวเข้าด้วยกัน "
        "และระบุประเภทของพารามิเตอร์ที่ใช้ครับ"
    )
    assert is_repetitive_response(normal_code_response) is False, "❌ FAIL: คำตอบปกติไม่ควรถูก Flag!"
    print("✅ PASS: คำตอบปกติที่มีโค้ดซ้ำเล็กน้อยไม่ถูก Flag ผิด\n")

    print("=" * 70)
    print("TEST: Weighted Router (Stateless)")
    print("=" * 70)
    for q in ["เขียน Python function ที่บวกเลขสองตัว", "def foo(): pass"]:
        result = route_query(q)
        print(f"Query: {q!r} -> {result}\n")

    print("=" * 70)
    print("TEST: Router — History Continuation")
    print("=" * 70)
    r1 = route_query("แก้ให้หน่อย", previous_expert="python")
    assert r1["expert"] == "python" and r1["method"] == "history_continuation"
    print("✅ PASS: 'แก้ให้หน่อย' ใช้ previous_expert ต่อ\n")

    r2 = route_query("อธิบายเพิ่มอีกหน่อย", previous_expert="python")
    assert r2["expert"] == "python", "❌ FAIL: ควรใช้ previous_expert ต่อ!"
    print("✅ PASS: 'อธิบายเพิ่มอีกหน่อย' ไม่ถูกเข้าใจผิดว่าเป็นการลาก่อนแล้ว\n")

    r2b = route_query("อธิบายว่าแก้ตรงไหน", previous_expert="python")
    assert r2b["expert"] == "python", "❌ FAIL: Bug 'บาย' ใน 'อธิบาย' กลับมาอีกแล้ว!"
    print("✅ PASS: 'อธิบายว่าแก้ตรงไหน' ไม่ถูก Flag ผิดเป็น Greeting แล้ว\n")

    r3 = route_query("แก้ให้หน่อย", previous_expert=None)
    assert r3["expert"] is None
    print("✅ PASS: ไม่มี previous_expert ยังทำงานแบบเดิม\n")

    r4 = route_query("ขอบคุณครับ", previous_expert="typescript")
    assert r4["expert"] is None, "❌ FAIL: คำขอบคุณจริงๆต้องยัง Flag เป็น Greeting!"
    print("✅ PASS: 'ขอบคุณครับ' ยังถูก Flag เป็น Greeting ถูกต้อง\n")

    r5 = route_query("บาย", previous_expert="python")
    assert r5["expert"] is None, "❌ FAIL: คำว่า 'บาย' โดดๆต้องยัง Flag เป็น Greeting!"
    print("✅ PASS: 'บาย' คำเดียวโดดๆ ยังถูก Flag เป็น Greeting ถูกต้อง\n")

    print("=" * 70)
    print("TEST: is_code_refusal")
    print("=" * 70)
    assert is_code_refusal("ขออภัยครับ ผมไม่สามารถช่วยคุณเขียนโค้ด Python ที่บวกเลขสองตัวได้ครับ") is True
    assert is_code_refusal("```python\ndef add(a, b):\n    return a + b\n```\nครับ") is False
    print("✅ PASS: Code Refusal Detector ทำงานถูกต้อง\n")

    print("=" * 70)
    print("TEST: Language purity filter — Emoji")
    print("=" * 70)
    result_emoji = clean_or_flag_response("ผมไม่สามารถทำให้🐍🐍🐍เป็นคำว่า python ได้ครับ 😊👍")
    assert result_emoji["is_clean"] is True
    print("✅ PASS: Emoji ไม่ถูก Flag ผิด\n")

    print("=" * 70)
    print("TEST: Persona enforcer")
    print("=" * 70)
    fixed = enforce_persona("ขออภัยค่ะ ฉันไม่เข้าใจคำถามนี้ค่ะ")
    assert "ค่ะ" not in fixed and "ฉัน" not in fixed
    print("✅ PASS\n")

    print("=" * 70)
    print("🎉 SELF-TEST ทั้งหมดผ่านเรียบร้อย")
    print("=" * 70)