src/agents/refusals.py

"""Canned, language-matched refusals for the guardrail layer.

Centralized so the three refusal sites — the `out_of_scope` router branch, the
input-guard block, and the content-filter fallback — speak with one voice.

Deterministic on purpose (no LLM): the refusal must not call the same model that
just refused or failed. Language is picked with a tiny marker check (EN default,
ID when Indonesian markers are present), mirroring the lightweight detector in
`handlers/help.py` — enough for a one-line reply without an LLM round-trip.
"""

from __future__ import annotations

import re

# Subset of help.py's ID marker set — enough to pick EN vs ID for a short refusal.
# Includes the common Indonesian injection verbs ("lupakan"=forget, "abaikan"=ignore)
# so an ID-language jailbreak still gets an ID-language refusal.
_ID_MARKERS = frozenset({
    "yang", "dan", "apa", "gimana", "bagaimana", "kenapa", "mengapa", "aku", "saya",
    "tolong", "ini", "itu", "untuk", "dengan", "tidak", "nggak", "enggak", "bisa",
    "mau", "buat", "dari", "kamu", "berapa", "kapan", "siapa", "dimana", "adalah",
    "akan", "sudah", "belum", "lupakan", "abaikan", "kredensial", "tunjukkan",
})


def _is_indonesian(message: str) -> bool:
    tokens = re.findall(r"[a-z']+", (message or "").lower())
    return any(t in _ID_MARKERS for t in tokens)


# Off-topic / out-of-scope: the request is benign but outside what the assistant does.
_OUT_OF_SCOPE = {
    "en": (
        "That's outside what I can help with — I'm a data assistant, so I can only work "
        "with the sources you've connected to Data Eyond. Ask me a question about your "
        "data, or type /help to see what I can do."
    ),
    "id": (
        "Itu di luar yang bisa saya bantu — saya asisten data, jadi saya hanya bisa "
        "bekerja dengan sumber data yang Anda hubungkan ke Data Eyond. Ajukan pertanyaan "
        "tentang data Anda, atau ketik /help untuk melihat yang bisa saya lakukan."
    ),
}

# Blocked: a manipulation / injection / secret-extraction / abuse attempt.
_BLOCKED = {
    "en": (
        "I can't help with that request. I'm here to analyze the data you've connected — "
        "ask me a question about your data and I'll take it from there."
    ),
    "id": (
        "Saya tidak bisa membantu permintaan tersebut. Saya di sini untuk menganalisis "
        "data yang Anda hubungkan — ajukan pertanyaan tentang data Anda dan saya bantu."
    ),
}


# Data-gap: the planner judged the bound sources cannot answer the question
# (planner.md "When the catalog cannot answer"). Deterministic wrapper on
# purpose — the model that declined to plan is not re-asked to prose it up.
# Keyed on the pipeline's reply_language ("Indonesian"/"English"), not marker
# detection: the upstream language decision is authoritative here.
_DATA_GAP = {
    "en": (
        "I can't answer that from the data sources connected to this analysis. "
        "{reason}You can bind a source that holds this data, or ask me what's "
        "available (try /help or \"what data do I have?\")."
    ),
    "id": (
        "Saya tidak bisa menjawab itu dari sumber data yang terhubung ke "
        "analisis ini. {reason}Anda bisa menambahkan sumber yang memuat data "
        "tersebut, atau tanyakan data apa yang tersedia (coba /help atau "
        "\"data apa yang saya punya?\")."
    ),
}


def data_gap_message(reason: str | None, reply_language: str | None = None) -> str:
    """Answer for an infeasible analysis: the bound sources lack the asked-for data.

    `reason` is the planner's `infeasible_reason` (may be None/empty);
    `reply_language` is the pipeline's detected language ("Indonesian"/"English").
    """
    detail = (reason or "").strip()
    if detail and not detail.endswith((".", "!", "?")):
        detail += "."
    lang = "id" if reply_language == "Indonesian" else "en"
    return _DATA_GAP[lang].format(reason=f"{detail} " if detail else "")


def out_of_scope_message(message: str) -> str:
    """Refusal for a benign but out-of-scope request (the `out_of_scope` intent)."""
    return _OUT_OF_SCOPE["id" if _is_indonesian(message) else "en"]


def blocked_message(message: str) -> str:
    """Refusal for a blocked request (injection / secrets / abuse / content-filter)."""
    return _BLOCKED["id" if _is_indonesian(message) else "en"]