"""Short-term conversation state for the assistant channel.

Two separate concerns live here, and keeping them apart is the point:

* **Working memory** — the last few turns the model needs to follow a thread.
  Redis, TTL'd, per (person, chat). Losing it costs context, never data.
* **Isolation** — one person's history is stored under their own server-resolved
  person id, so another employee's turns are not merely filtered out of the
  prompt, they are never loaded on that code path at all.

The durable transcript is a different thing entirely and lives in the database
(:mod:`app.assistant.models`). This module can be flushed at any time.
"""
from __future__ import annotations

import json
import logging
import uuid
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

from app.assistant.settings import (
    DEFAULT_HISTORY_TURNS,
    DEFAULT_SESSION_TTL_MINUTES,
    SETTING_HISTORY_TURNS,
    SETTING_SESSION_TTL_MINUTES,
    bounded_int_setting,
)
from app.core.redis_client import get_redis

logger = logging.getLogger(__name__)

_HISTORY_PREFIX = "assistant:history:"
_LOCK_PREFIX = "assistant:lock:"
#: Bumped by every reset. A turn captures it when it loads the thread and its
#: write is refused if the value moved — see :func:`clear_history`.
_EPOCH_PREFIX = "assistant:epoch:"
_ACTIVE_CHAT_PREFIX = "assistant:active_chat:"
#: The epoch must outlive the HISTORY it guards, or the guard expires first and
#: leaves two holes: a turn whose captured epoch has since vanished has its write
#: refused for nothing, and a second reset re-creating the key from zero can hand
#: out a value an in-flight turn already holds (``INCR`` on a missing key returns
#: 1 again), so that turn's stale write is waved through. Ten times the history
#: ceiling, which is itself capped at 1440 minutes by ``bounded_int_setting``.
_EPOCH_TTL_SECONDS = 1440 * 60 * 10
#: Must outlast the worst-case turn, or the lock evaporates under a running one
#: and message 2 proceeds concurrently — which would give both turns the same
#: ``turn_count`` and so the same message ``seq``, corrupting the transcript
#: order the explicit sequence exists to guarantee. The kernel's own turn budget
#: is 110s and a final hop can add 60s, plus unbounded bridged-tool time, so
#: this is set well above that rather than snugly against it. A worker that dies
#: still frees the chat within this window.
_LOCK_TTL_SECONDS = 600

#: Typing these resets the thread. Kept literal and multilingual rather than
#: model-detected: a reset the employee asked for must never depend on the model
#: being available or agreeing.
RESET_COMMANDS = frozenset({
    "新话题", "新对话", "重置", "清空",
    "/new", "/reset", "new topic", "reset",
    "新しい話題", "リセット",
})

# Deterministic channel controls stay outside the LLM: employees must be able
# to discover what Helix can do even when the model is unavailable.
HELP_COMMANDS = frozenset({
    "/help", "help", "帮助", "使用帮助", "ヘルプ",
})


def is_reset_command(text: str) -> bool:
    return text.strip().lower() in {c.lower() for c in RESET_COMMANDS}


def is_help_command(text: str) -> bool:
    return text.strip().lower() in {c.lower() for c in HELP_COMMANDS}


def _history_key(person_id: Any, chat_id: str) -> str:
    return f"{_HISTORY_PREFIX}{person_id}:{chat_id}"


def _epoch_key(person_id: Any, chat_id: str) -> str:
    return f"{_EPOCH_PREFIX}{person_id}:{chat_id}"


def _active_chat_key(person_id: Any) -> str:
    return f"{_ACTIVE_CHAT_PREFIX}{person_id}"


async def remember_active_chat(*, person_id: Any, chat_id: str) -> None:
    """Remember the latest direct-chat key for a menu-triggered reset.

    Bot-menu events identify the employee but do not carry the P2P chat id.
    Keeping this short-lived pointer lets "新话题" clear the same conversation
    the employee was just using, without broad key scans or clearing another
    chat.
    """
    if not chat_id:
        return
    try:
        await get_redis().set(
            _active_chat_key(person_id), chat_id,
            ex=DEFAULT_SESSION_TTL_MINUTES * 60,
        )
    except Exception as exc:  # noqa: BLE001
        logger.warning("assistant: active chat pointer save failed: %s", exc)


async def active_chat(*, person_id: Any) -> str | None:
    try:
        value = await get_redis().get(_active_chat_key(person_id))
    except Exception as exc:  # noqa: BLE001
        logger.warning("assistant: active chat pointer read failed: %s", exc)
        return None
    if value is None:
        return None
    return value.decode() if isinstance(value, bytes) else str(value)


#: 「我要找人工,但还没说是什么事」。帮助卡上的那颗按钮在对话还是空的时候会
#: 记一笔,员工的下一句话就直接做成工单,不再经过模型。
_HANDOFF_ARMED_PREFIX = "assistant:handoff_armed:"
#: 15 分钟。够一个人打完一句话,又短到不会在半小时后把一句无关的问候变成工单。
_HANDOFF_ARMED_TTL_SECONDS = 15 * 60


def _handoff_armed_key(person_id: Any, chat_id: str) -> str:
    return f"{_HANDOFF_ARMED_PREFIX}{person_id}:{chat_id}"


async def arm_handoff(*, person_id: Any, chat_id: str) -> bool:
    """记下「这个人要找人工,正在补一句说明」。Redis 不可用时返回 ``False``。

    返回值是给调用方用来**说实话**的:记不下这一笔时,员工那句说明会被当成
    普通问题交给模型,所以那种情况下的措辞必须不同 —— 不能承诺一件做不到的事。
    """
    if not chat_id:
        return False
    try:
        await get_redis().set(
            _handoff_armed_key(person_id, chat_id), "1",
            ex=_HANDOFF_ARMED_TTL_SECONDS,
        )
    except Exception as exc:  # noqa: BLE001 — degrade, never refuse
        logger.warning("assistant: handoff arm failed: %s", exc)
        return False
    return True


async def take_handoff_armed(*, person_id: Any, chat_id: str) -> bool:
    """这句话是不是那句补充说明?**取走**标记,一次只兑现一条。

    ``GETDEL`` 而不是「读了再删」:两条消息同时到达时,读-删之间的窗口会让两条
    都被当成说明,于是同一件事开出两张工单。
    """
    if not chat_id:
        return False
    try:
        taken = await get_redis().getdel(_handoff_armed_key(person_id, chat_id))
    except Exception as exc:  # noqa: BLE001 — degrade, never refuse
        logger.warning("assistant: handoff flag read failed: %s", exc)
        return False
    return bool(taken)


#: 这个人**在这个会话里**上一次是用哪种语言说话的。
#:
#: 存在这里的理由和 history 一样:它是会话状态,不是这个人的属性。他的语言
#: 偏好由 ``identity_persons.preferred_language`` 说了算,而这一格只回答「刚才
#: 那几句是用什么语言问的」—— 于是一句「ok」或者一个裸资产号不会把一段英文
#: 对话翻回中文。丢了只是退回偏好,不丢数据。
_LANG_PREFIX = "assistant:lang:"


def _lang_key(person_id: Any, chat_id: str) -> str:
    return f"{_LANG_PREFIX}{person_id}:{chat_id}"


async def remember_lang(*, person_id: Any, chat_id: str, lang: str) -> None:
    """记下这一轮用的语言。TTL 与线程同长 —— 话题过期了,语言也不该赖着。"""
    if not chat_id or not lang:
        return
    try:
        await get_redis().set(
            _lang_key(person_id, chat_id), lang,
            ex=DEFAULT_SESSION_TTL_MINUTES * 60,
        )
    except Exception as exc:  # noqa: BLE001 — degrade, never refuse
        logger.warning("assistant: language memo save failed: %s", exc)


async def forget_lang(*, person_id: Any, chat_id: str) -> None:
    """丢掉这个会话的语言记忆。

    只有一个调用者:员工**明确**选了一种语言的时候。那一格记的是「他刚才用什么
    语言说话」,而他刚刚说了要用哪种 —— 不清掉的话,下一句「ok」会把他拉回旧
    语言,而他刚按过那颗按钮,只会觉得开关坏了。
    """
    if not chat_id:
        return
    try:
        await get_redis().delete(_lang_key(person_id, chat_id))
    except Exception as exc:  # noqa: BLE001 — degrade, never refuse
        logger.warning("assistant: language memo clear failed: %s", exc)


async def recall_lang(*, person_id: Any, chat_id: str) -> str | None:
    """上一轮用的语言,读不到就 ``None``(调用方退回这个人的偏好)。"""
    if not chat_id:
        return None
    try:
        value = await get_redis().get(_lang_key(person_id, chat_id))
    except Exception as exc:  # noqa: BLE001 — degrade, never refuse
        logger.warning("assistant: language memo read failed: %s", exc)
        return None
    if value is None:
        return None
    return value.decode() if isinstance(value, bytes) else str(value)


async def current_epoch(*, person_id: Any, chat_id: str) -> str | None:
    """The thread's reset counter as it stands now, or None if unreadable.

    Captured at the START of a turn and handed back to :func:`save_history`, so a
    reset that happens while the turn is running cannot be silently undone by
    that turn's own write. ``None`` (Redis unreadable) degrades to an
    unconditional write — the same trade every other Redis path here makes.
    """
    try:
        value = await get_redis().get(_epoch_key(person_id, chat_id))
    except Exception as exc:  # noqa: BLE001 — degrade, never refuse
        logger.warning("assistant: history epoch unavailable: %s", exc)
        return None
    if value is None:
        return "0"
    return value.decode() if isinstance(value, bytes) else str(value)


async def load_history(
    db: AsyncSession, *, person_id: Any, chat_id: str,
) -> list[dict[str, str]]:
    """Recent turns for this person in this chat, oldest first.

    Redis failure degrades to an empty history — the employee gets a
    context-free but correct answer, which beats refusing to answer.
    """
    try:
        raw = await get_redis().get(_history_key(person_id, chat_id))
    except Exception as exc:  # noqa: BLE001 — degrade, never refuse
        logger.warning("assistant: history unavailable: %s", exc)
        return []
    if not raw:
        return []
    try:
        parsed = json.loads(raw)
    except (ValueError, TypeError):
        return []
    if not isinstance(parsed, list):
        return []
    return [
        {"role": str(t.get("role")), "content": str(t.get("content"))}
        for t in parsed
        if isinstance(t, dict)
        and t.get("role") in ("user", "assistant")
        and str(t.get("content") or "").strip()
    ]


#: Write the thread only if the reset counter still reads what the turn saw.
#:
#: A reset is the ONE control an employee has over the assistant's memory, and it
#: was possible to take it away from them without a trace: 「新话题」 during a
#: 15-second turn deleted the history, answered 「好的，我们从头开始」, and then the
#: still-running turn wrote ``[*prior, Q, A]`` back over the cleared key. The
#: employee was told the thread was reset and the model kept every word of it.
_SAVE_IF_EPOCH = """
if ARGV[3] == '' or redis.call('GET', KEYS[2]) == ARGV[3]
   or (ARGV[3] == '0' and redis.call('EXISTS', KEYS[2]) == 0) then
  redis.call('SET', KEYS[1], ARGV[1], 'EX', ARGV[2])
  return 1
end
return 0
"""


#: Clear the thread and bump its epoch in ONE step. ``INCR`` first inside the
#: script so a concurrent reader can never observe the new epoch with the old
#: history (or the reverse).
_CLEAR_AND_BUMP = """
redis.call('INCR', KEYS[2])
redis.call('EXPIRE', KEYS[2], ARGV[1])
redis.call('DEL', KEYS[1])
return 1
"""


async def save_history(
    db: AsyncSession,
    *,
    person_id: Any,
    chat_id: str,
    history: list[dict[str, str]],
    expect_epoch: str | None = None,
) -> None:
    """Persist the trimmed thread. Best-effort: a lost write costs context only.

    ``expect_epoch`` is the value :func:`current_epoch` returned when this turn
    started. Pass it and the write becomes conditional; omit it and the write is
    unconditional (the escalation path, which never resets).
    """
    turns = await bounded_int_setting(
        db, SETTING_HISTORY_TURNS,
        default=DEFAULT_HISTORY_TURNS, minimum=2, maximum=40,
    )
    ttl_minutes = await bounded_int_setting(
        db, SETTING_SESSION_TTL_MINUTES,
        default=DEFAULT_SESSION_TTL_MINUTES, minimum=5, maximum=1440,
    )
    # Keep whole exchanges: an odd trim can strand an assistant reply as the
    # first entry, which reads to the model as if it spoke unprompted.
    trimmed = history[-(turns * 2):]
    try:
        wrote = await get_redis().eval(
            _SAVE_IF_EPOCH,
            2,
            _history_key(person_id, chat_id),
            _epoch_key(person_id, chat_id),
            json.dumps(trimmed, ensure_ascii=False),
            str(ttl_minutes * 60),
            "" if expect_epoch is None else expect_epoch,
        )
        if not wrote:
            logger.info(
                "assistant: history write dropped — the thread was reset mid-turn",
            )
    except Exception as exc:  # noqa: BLE001
        logger.warning("assistant: history save failed: %s", exc)


async def clear_history(*, person_id: Any, chat_id: str) -> bool:
    """Forget the thread so nothing can bring it back. False if it did not stick.

    One script, three reasons. **Atomic**: as three round trips, a turn starting
    between the ``INCR`` and the ``DEL`` read the NEW epoch and the OLD history,
    so it answered with the memory the employee had just cleared and its
    write-back passed the CAS — the defect the epoch exists to prevent, arriving
    through the gap in its own implementation. **All-or-nothing**: a partial
    failure used to be swallowed while the caller still said 「好的，我们从头
    开始」. **Ordered**: the bump lands with the delete, so no reader can see one
    without the other.

    The returned flag is what lets the caller stop claiming a reset it did not
    get.
    """
    try:
        return bool(await get_redis().eval(
            _CLEAR_AND_BUMP, 2,
            _history_key(person_id, chat_id),
            _epoch_key(person_id, chat_id),
            str(_EPOCH_TTL_SECONDS),
        ))
    except Exception as exc:  # noqa: BLE001
        logger.warning("assistant: history clear failed: %s", exc)
        return False


#: Release only if we still hold it. A plain DELETE would let a turn that
#: overran the TTL free the NEXT turn's lock on its way out, so a third message
#: would then run concurrently with the second — the failure mode the lock
#: exists to prevent, arriving one turn later.
_RELEASE_IF_OWNER = """
if redis.call('GET', KEYS[1]) == ARGV[1] then
  return redis.call('DEL', KEYS[1])
end
return 0
"""


async def acquire_turn_lock(*, person_id: Any, chat_id: str) -> str | None:
    """Serialize turns within one chat.

    Without this, two messages sent a second apart run concurrently, read the
    same history and reply out of order. Returns an owner token to pass to
    :func:`release_turn_lock`, or ``None`` when another turn holds the chat.

    Fail-open on Redis fault: answering twice out of order is a worse day than
    not answering at all, but only just.
    """
    owner = uuid.uuid4().hex
    try:
        claimed = await get_redis().set(
            f"{_LOCK_PREFIX}{person_id}:{chat_id}", owner,
            nx=True, ex=_LOCK_TTL_SECONDS,
        )
    except Exception as exc:  # noqa: BLE001
        logger.warning("assistant: turn lock unavailable: %s", exc)
        return owner
    return owner if claimed else None


async def release_turn_lock(
    *, person_id: Any, chat_id: str, owner: str | None,
) -> None:
    if not owner:
        return
    try:
        await get_redis().eval(
            _RELEASE_IF_OWNER, 1, f"{_LOCK_PREFIX}{person_id}:{chat_id}", owner,
        )
    except Exception as exc:  # noqa: BLE001
        logger.warning("assistant: turn lock release failed: %s", exc)