"""Enforce user-specified delegate role on Lead projects (override docs→content drift)."""

from __future__ import annotations

import logging
import re
from typing import TYPE_CHECKING

from butler.utilities.env_parse import env_truthy

if TYPE_CHECKING:
    from butler.tools.delegate_phases import DelegateRunState

logger = logging.getLogger(__name__)

_DEV_SIGNALS: tuple[re.Pattern[str], ...] = (
    re.compile(r"role\s*[=:]\s*dev\b", re.I),
    re.compile(r"委派\s*开发", re.I),
    re.compile(r"开发代理", re.I),
    re.compile(r"交给\s*开发", re.I),
    re.compile(r"禁止(用|使用)?\s*content", re.I),
    re.compile(r"不要(用|使用)?\s*content", re.I),
    re.compile(r"禁止\s*内容代理", re.I),
)

_CONTENT_SIGNALS: tuple[re.Pattern[str], ...] = (
    re.compile(r"role\s*[=:]\s*content\b", re.I),
    re.compile(r"委派\s*内容", re.I),
    re.compile(r"内容代理", re.I),
    re.compile(r"交给\s*content", re.I),
    re.compile(r"禁止(用|使用)?\s*dev\b", re.I),
    re.compile(r"不要(用|使用)?\s*dev\b", re.I),
    re.compile(r"禁止\s*开发代理", re.I),
)

_FORBID_DELEGATE_SIGNALS: tuple[re.Pattern[str], ...] = (
    re.compile(r"不要委派", re.I),
    re.compile(r"禁止委派", re.I),
    re.compile(r"不(要|许)委派", re.I),
    re.compile(r"不要\s*delegate", re.I),
    re.compile(r"勿\s*委派", re.I),
)

_READONLY_ONLY_SIGNALS: tuple[re.Pattern[str], ...] = (
    re.compile(r"只读", re.I),
    re.compile(r"仅查看", re.I),
    re.compile(r"不要改", re.I),
    re.compile(r"不要修改", re.I),
    re.compile(r"勿改", re.I),
    re.compile(r"read[\s-]?only", re.I),
)

_DELEGATE_INTENT_SIGNALS: tuple[re.Pattern[str], ...] = (
    re.compile(r"委派", re.I),
    re.compile(r"delegate", re.I),
    re.compile(r"交给\s*(开发|内容|review|审查)?\s*代理", re.I),
    re.compile(r"开发代理", re.I),
    re.compile(r"内容代理", re.I),
)

_LEAD_SELF_READ_FILE = re.compile(
    r"(自己|请你|请你先|本线程).{0,12}read_file|read_file.{0,40}(后|然后)",
    re.I,
)


def _normalize_role(role: str) -> str:
    return str(role or "").replace("_agent", "").strip().lower()


def _user_query_head(text: str) -> str:
    """Parse only the user-facing lines; ignore trailing Skill/Memory injections."""
    raw = str(text or "").strip()
    if not raw:
        return ""
    for marker in (
        "## 相关知识",
        "## Related knowledge",
        "### `",
        "<!-- butler-skill",
    ):
        idx = raw.find(marker)
        if idx > 0:
            return raw[:idx].strip()
    return raw


def user_explicit_delegate_role(user_text: str) -> str | None:
    """Return ``dev`` / ``content`` / ``review`` when user pinned a role, else None."""
    text = _user_query_head(user_text)
    if not text:
        return None
    dev_hit = any(p.search(text) for p in _DEV_SIGNALS)
    content_hit = any(p.search(text) for p in _CONTENT_SIGNALS)
    # Explicit role= params in user head — dev before review (Skill blocks may mention review)
    if re.search(r"role\s*[=:]\s*dev\b", text, re.I):
        return "dev"
    if re.search(r"role\s*[=:]\s*content\b", text, re.I):
        return "content"
    if re.search(r"role\s*[=:]\s*review\b", text, re.I):
        return "review"
    if dev_hit and not content_hit:
        return "dev"
    if content_hit and not dev_hit:
        return "content"
    return None


def _turn_user_text() -> str:
    from butler.tools.network_search_policy import _turn_user_query

    text = _turn_user_query()
    if text:
        return str(text)
    from butler.execution_context import get_current_session_key
    from butler.tools.delegate_role_guard_ops import turn_user_text_from_epoch_safe

    sk = str(get_current_session_key() or "").strip()
    if sk:
        query = turn_user_text_from_epoch_safe(sk)
        if query:
            return str(query)
    return ""


def _explicit_role_from_state(state: "DelegateRunState") -> str | None:
    blob = "\n".join(
        part
        for part in (
            state.task or "",
            state.context or "",
            state.original_context or "",
        )
        if part
    )
    return user_explicit_delegate_role(blob)


def _is_lead_turn() -> bool:
    from butler.tools.delegate_role_guard_ops import is_lead_project_turn_safe

    lead = is_lead_project_turn_safe()
    return bool(lead) if lead is not None else False


def lead_readonly_gate_enabled() -> bool:
    """Block Lead mistaken delegate_task on read-only intents (default on)."""
    return bool(env_truthy("BUTLER_LEAD_READONLY_GATE", default=True))


def lead_forbids_delegate(user_text: str) -> bool:
    """True when Lead must answer read-only (no delegate_task) from user intent."""
    text = _user_query_head(user_text)
    if not text:
        return False
    if user_explicit_delegate_role(text):
        return False
    if any(p.search(text) for p in _FORBID_DELEGATE_SIGNALS):
        return True
    if any(p.search(text) for p in _READONLY_ONLY_SIGNALS):
        if any(p.search(text) for p in _DELEGATE_INTENT_SIGNALS):
            return False
        return True
    lowered = text.lower()
    factory_ctx = (
        "workflow_state" in lowered
        or "novel-factory" in lowered
        or "流水线" in text
    )
    status_ask = any(
        token in text
        for token in ("阶段", "进度", "厂情", "step", "phase", "STEP", "PHASE")
    )
    if factory_ctx and status_ask and _LEAD_SELF_READ_FILE.search(text):
        return True
    return False


def block_lead_readonly_delegate(*, depth: int = 0) -> str | None:
    """Return tool-error JSON when Lead turn must not delegate; else None."""
    if not lead_readonly_gate_enabled():
        return None
    if int(depth or 0) > 0:
        return None
    if not _is_lead_turn():
        return None
    if not lead_forbids_delegate(_turn_user_text()):
        return None
    import json

    return json.dumps(
        {
            "error": (
                "Lead 只读厂情：请本线程 read_file novel-factory/workflow_state.json "
                "或 run_workflow(novel-factory-status)，勿 delegate_task"
            ),
            "code": "LEAD_READONLY_NO_DELEGATE",
            "hint": "用户要求只读或不委派时，Lead 不得 delegate_task。",
        },
        ensure_ascii=False,
    )


def apply_user_role_override(state: "DelegateRunState") -> bool:
    """If user pinned role=dev/content on a Lead turn, override model-chosen role."""
    if int(getattr(state, "depth", 0) or 0) > 0:
        return False
    if not _is_lead_turn():
        return False
    explicit = user_explicit_delegate_role(_turn_user_text())
    if not explicit:
        explicit = _explicit_role_from_state(state)
    if not explicit:
        return False
    current = _normalize_role(state.role)
    if current == explicit:
        return False
    logger.info(
        "Lead delegate role override: %s -> %s (user explicit)",
        current or "?",
        explicit,
    )
    state.role = explicit
    note = (
        f"[role_override] 用户明确要求 role={explicit}；"
        f"已覆盖 Lead 默认路由（原 role={current or '?'}）。"
    )
    ctx = str(state.context or "").strip()
    state.context = f"{ctx}\n\n{note}".strip() if ctx else note
    return True


__all__ = [
    "apply_user_role_override",
    "block_lead_readonly_delegate",
    "lead_forbids_delegate",
    "lead_readonly_gate_enabled",
    "user_explicit_delegate_role",
]