"""Pending council offers: storage, approve/dismiss, HTTP helpers."""

from __future__ import annotations

import logging
from typing import Any

from utils.council_triggers import (
    COUNCIL_OFFER_FAILED_TAIL,
    COUNCIL_OFFER_PROMPT_TAIL,
    COUNCIL_PHRASE_HINT_TAIL,
    COUNCIL_SKIP_LINE,
    CouncilTurnAction,
    build_captain_frame,
    frame_hash,
    is_council_foreground_allowed,
    new_offer_id,
    thread_agent_running,
    evaluate_council_turn,
)
_log = logging.getLogger(__name__)

OFFER_TTL_SECONDS = 600


def create_pending_offer(
    *,
    thread_id: str,
    council_frame: str,
    rationale: str = "",
) -> dict[str, Any]:
    from chat_storage import set_thread_council_pending_offer

    tid = (thread_id or "").strip()
    frame = build_captain_frame(council_frame)
    if not tid or not frame:
        raise ValueError("thread_id and council_frame are required")
    offer_id = new_offer_id()
    set_thread_council_pending_offer(
        tid,
        offer_id=offer_id,
        council_frame=frame,
        frame_hash=frame_hash(frame),
        rationale=(rationale or "").strip(),
    )
    return {
        "offer_id": offer_id,
        "thread_id": tid,
        "council_frame": frame,
        "rationale": (rationale or "").strip(),
    }


def get_pending_offer_for_thread(thread_id: str) -> dict[str, Any] | None:
    """Read-only: return the live pending offer, or None if absent/expired.

    Must stay side-effect-free — this is called from ``GET /chat/thread_state`` (banner
    refresh) on every poll. Deleting the expired row here was a read-path-write seam that
    raced approve/dismiss and could drop an offer mid-refresh. Expired rows are reaped on
    the write paths instead (``approve_pending_offer`` / ``dismiss_pending_offer``), and are
    harmless to leave in place because they are filtered out as expired on every read.
    """
    from chat_storage import get_thread_council_pending_offer

    tid = (thread_id or "").strip()
    if not tid:
        return None
    row = get_thread_council_pending_offer(tid)
    if not row:
        return None
    if _offer_expired(row):
        return None
    return row


def clear_pending_offer(thread_id: str) -> None:
    from chat_storage import clear_thread_council_pending_offer

    clear_thread_council_pending_offer((thread_id or "").strip())


def _offer_expired(row: dict[str, Any]) -> bool:
    try:
        from chat_storage import council_offer_age_seconds

        age = council_offer_age_seconds(row.get("created_at") or "")
        return age > OFFER_TTL_SECONDS
    except Exception:
        _log.debug("council offer expiry check failed", exc_info=True)
        return True


def approve_pending_offer(
    *,
    thread_id: str,
    offer_id: str,
) -> dict[str, Any]:
    """Spawn council batch from a pending offer. Caller handles agent wake."""
    from utils.council_orchestrator import (
        council_batch_created_user_message,
        probe_council_active,
        spawn_council_batch,
        spawn_preflight_error,
    )

    tid = (thread_id or "").strip()
    oid = (offer_id or "").strip()
    if not tid or not oid:
        return {"ok": False, "error": "thread_id and offer_id are required", "status": 400}

    row = get_pending_offer_for_thread(tid)
    if not row:
        # Reap any expired/absent row on this write path (the read path no longer clears).
        try:
            clear_pending_offer(tid)
        except Exception:
            _log.debug("expired offer reap failed", exc_info=True)
        return {"ok": False, "error": "offer expired or not found", "status": 410}
    if str(row.get("offer_id") or "").strip() != oid:
        return {"ok": False, "error": "offer_id does not match this thread", "status": 409}

    if thread_agent_running(tid):
        return {"ok": False, "error": "thread agent is still running; try again shortly", "status": 409}

    if probe_council_active(tid):
        clear_pending_offer(tid)
        return {"ok": False, "error": "council already active for this thread", "status": 409}

    preflight = spawn_preflight_error(origin_thread_id=tid)
    if preflight:
        deliver_council_spawn_refusal(tid, reason=preflight)
        return {"ok": False, "error": preflight, "status": 400}

    frame = str(row.get("council_frame") or "").strip()
    try:
        result = spawn_council_batch(origin_thread_id=tid, council_frame=frame)
    except ValueError as exc:
        err = str(exc).strip() or "preflight refused"
        deliver_council_spawn_refusal(tid, reason=err)
        return {"ok": False, "error": err, "status": 400}
    except Exception as exc:
        _log.warning("approve_pending_offer spawn failed", exc_info=True)
        err = f"spawn failed: {type(exc).__name__}"
        deliver_council_spawn_refusal(tid, reason=err)
        return {"ok": False, "error": err, "status": 500}

    clear_pending_offer(tid)
    bid = str(result.get("council_batch_id") or "")
    msg = str(result.get("message") or council_batch_created_user_message(bid))
    return {
        "ok": True,
        "council_batch_id": bid,
        "message": msg,
        "honest_stop": True,
    }


def dismiss_pending_offer(*, thread_id: str, offer_id: str) -> dict[str, Any]:
    """Clear offer and persist skip line for single-pass continuation."""
    tid = (thread_id or "").strip()
    oid = (offer_id or "").strip()
    if not tid or not oid:
        return {"ok": False, "error": "thread_id and offer_id are required", "status": 400}

    row = get_pending_offer_for_thread(tid)
    if not row:
        clear_pending_offer(tid)
        return {"ok": True, "already_cleared": True}
    if str(row.get("offer_id") or "").strip() != oid:
        return {"ok": False, "error": "offer_id does not match this thread", "status": 409}

    clear_pending_offer(tid)
    try:
        from chat_storage import append_messages

        append_messages(tid, [{"role": "user", "content": COUNCIL_SKIP_LINE}])
    except Exception:
        _log.warning("dismiss offer append failed", exc_info=True)
        return {"ok": False, "error": "failed to record skip preference", "status": 500}

    return {"ok": True, "skip_line": COUNCIL_SKIP_LINE}


def build_council_honest_stop_prompt(thread_id: str) -> str:
    """Honest-stop user inject after council spawn (approve or always tier)."""
    from agent_loop_guardrails import _format_council_active_message
    from utils.council_orchestrator import (
        format_council_status_snapshot,
        probe_council_batch_id,
        tasks_for_council_batch,
    )

    tid = (thread_id or "").strip()
    bid = probe_council_batch_id(tid)
    snap = format_council_status_snapshot(tasks_for_council_batch(tid, bid))
    return _format_council_active_message(snap)


def schedule_council_honest_stop_wake(thread_id: str) -> None:
    from utils.council_delivery import register_pending_council_wake

    tid = (thread_id or "").strip()
    if tid:
        register_pending_council_wake(tid, reason="council_honest_stop")


def schedule_council_skip_wake(thread_id: str) -> None:
    from utils.council_delivery import register_pending_council_wake

    tid = (thread_id or "").strip()
    if tid:
        register_pending_council_wake(tid, reason="council_skip")


def convene_council_via_slash(*, thread_id: str, council_frame: str) -> dict[str, Any]:
    """Spawn council from ``/council`` — bypasses Settings tier when mode is off."""
    from agent_loop_guardrails import _format_council_active_message
    from utils.council_orchestrator import (
        format_council_status_snapshot,
        spawn_council_batch,
        tasks_for_council_batch,
    )

    tid = (thread_id or "").strip()
    frame = build_captain_frame(council_frame)
    if not tid:
        raise ValueError("thread_id is required")
    if not frame:
        raise ValueError("question is required")
    result = spawn_council_batch(
        origin_thread_id=tid,
        council_frame=frame,
        allow_in_flight_turn=True,
        slash_override=True,
    )
    bid = str(result.get("council_batch_id") or "")
    msg = str(result.get("message") or "")
    # Slash handler persists the operator's `/council …` user bubble; skip synthetic
    # activation user row (tool/approve paths still use deliver_council_spawn_activation).
    try:
        from utils.council_delivery import notify_council_thread_state_refresh

        notify_council_thread_state_refresh(tid)
    except Exception:
        _log.debug("slash council spawn thread_state refresh failed", exc_info=True)
    snap = format_council_status_snapshot(tasks_for_council_batch(tid, bid))
    honest = _format_council_active_message(snap)
    schedule_council_honest_stop_wake(tid)
    return {
        "ok": True,
        "council_batch_id": bid,
        "message": honest,
    }


_SPAWN_REFUSAL_HEADING = "Council could not be convened."


def _spawn_refusal_already_persisted(thread_id: str, detail: str) -> bool:
    """Skip duplicate guard-marked refusal bubbles (e.g. repeated approve clicks)."""
    from utils.kanban_coordinator_delivery import GUARD_PREFIX

    tid = (thread_id or "").strip()
    needle = (detail or "").strip()[:120]
    if not tid:
        return False
    try:
        from chat_storage import get_messages

        checked = 0
        for msg in reversed(get_messages(tid) or []):
            if (msg.get("role") or "").strip() != "user":
                continue
            checked += 1
            content = str(msg.get("content") or "")
            if GUARD_PREFIX in content and _SPAWN_REFUSAL_HEADING in content:
                if not needle or needle in content:
                    return True
            if checked >= 8:
                break
    except Exception:
        _log.debug("spawn refusal dedupe probe failed", exc_info=True)
    return False


def deliver_council_spawn_refusal(thread_id: str, *, reason: str) -> bool:
    """Persist a guard-marked refusal bubble when council spawn is blocked."""
    from utils.kanban_coordinator_delivery import GUARD_PREFIX

    tid = (thread_id or "").strip()
    detail = (reason or "").strip()
    if not tid or not detail:
        return False
    if _spawn_refusal_already_persisted(tid, detail):
        try:
            from utils.council_delivery import (
                notify_council_chat_reload,
                notify_council_thread_state_refresh,
            )

            notify_council_chat_reload(tid)
            notify_council_thread_state_refresh(tid)
        except Exception:
            _log.debug("council spawn refusal dedupe reload failed", exc_info=True)
        return True
    content = "\n".join(
        (
            GUARD_PREFIX,
            _SPAWN_REFUSAL_HEADING,
            "",
            detail,
        )
    ).strip()
    try:
        from chat_storage import append_messages

        last_id = append_messages(tid, [{"role": "user", "content": content}])
        if last_id is None:
            return False
        try:
            from utils.council_delivery import (
                notify_council_chat_reload,
                notify_council_thread_state_refresh,
            )

            notify_council_chat_reload(tid)
            notify_council_thread_state_refresh(tid)
        except Exception:
            _log.debug("council spawn refusal chat reload failed", exc_info=True)
        return True
    except Exception:
        _log.warning("council spawn refusal append failed", exc_info=True)
        return False


def deliver_council_spawn_activation(thread_id: str, *, message: str) -> bool:
    """Persist activation user bubble after approve or always-tier spawn."""
    tid = (thread_id or "").strip()
    if not tid:
        return False
    content = (message or "").strip()
    if not content:
        return False
    try:
        from chat_storage import append_messages

        last_id = append_messages(tid, [{"role": "user", "content": content}])
        if last_id is None:
            return False
        try:
            from utils.council_delivery import (
                notify_council_chat_reload,
                notify_council_thread_state_refresh,
            )

            notify_council_chat_reload(tid)
            notify_council_thread_state_refresh(tid)
        except Exception:
            _log.debug("council activation reload failed", exc_info=True)
        return True
    except Exception:
        _log.warning("council activation append failed", exc_info=True)
        return False


def apply_council_turn_start(
    *,
    thread_id: str,
    user_text: str,
    policy_context: Any | None,
    subagent_mode: bool,
    is_heartbeat: bool,
) -> dict[str, Any]:
    """Run at foreground turn start; may create offer or auto-spawn."""
    out: dict[str, Any] = {
        "early_stop": False,
        "offer_created": False,
        "offer_failed": False,
        "offer_id": "",
        "prompt_tail": "",
        "message": "",
    }
    if not is_council_foreground_allowed(
        policy_context,
        subagent_mode=subagent_mode,
        is_heartbeat=is_heartbeat,
    ):
        return out

    decision = evaluate_council_turn(
        thread_id=thread_id,
        user_text=user_text,
        in_flight_turn=True,
    )

    if decision.action == CouncilTurnAction.SKIP:
        return out

    if decision.action == CouncilTurnAction.PHRASE_HINT:
        out["prompt_tail"] = COUNCIL_PHRASE_HINT_TAIL
        return out

    if decision.action == CouncilTurnAction.SPAWN_BLOCKED:
        from utils.council_triggers import COUNCIL_SPAWN_FAILED_TAIL

        reason = (decision.message or "").strip() or "Council could not be convened."
        deliver_council_spawn_refusal(thread_id, reason=reason)
        out["early_stop"] = True
        out["spawn_failed"] = True
        out["message"] = ""
        out["prompt_tail"] = COUNCIL_SPAWN_FAILED_TAIL.format(reason=reason)
        return out

    if decision.action == CouncilTurnAction.OFFER:
        try:
            offer = create_pending_offer(
                thread_id=thread_id,
                council_frame=decision.frame,
                rationale=decision.rationale,
            )
            out["offer_created"] = True
            out["offer_id"] = offer["offer_id"]
            out["prompt_tail"] = COUNCIL_OFFER_PROMPT_TAIL
        except Exception:
            _log.warning("council offer create failed", exc_info=True)
            out["offer_failed"] = True
            out["prompt_tail"] = COUNCIL_OFFER_FAILED_TAIL
        return out

    if decision.action == CouncilTurnAction.AUTO_SPAWN:
        from utils.council_orchestrator import (
            council_batch_created_user_message,
            format_council_status_snapshot,
            spawn_council_batch,
            tasks_for_council_batch,
        )
        from agent_loop_guardrails import _format_council_active_message

        from utils.council_triggers import COUNCIL_SPAWN_FAILED_TAIL

        try:
            result = spawn_council_batch(
                origin_thread_id=thread_id,
                council_frame=decision.frame,
                allow_in_flight_turn=True,
            )
            bid = str(result.get("council_batch_id") or "")
            msg = str(result.get("message") or council_batch_created_user_message(bid))
            deliver_council_spawn_activation(thread_id, message=msg)
            snap = format_council_status_snapshot(tasks_for_council_batch(thread_id, bid))
            out["early_stop"] = True
            out["message"] = _format_council_active_message(snap)
        except ValueError as exc:
            reason = str(exc).strip() or "preflight refused"
            _log.warning("council auto_spawn refused: %s", reason)
            deliver_council_spawn_refusal(thread_id, reason=reason)
            out["early_stop"] = True
            out["spawn_failed"] = True
            out["message"] = ""
            out["prompt_tail"] = COUNCIL_SPAWN_FAILED_TAIL.format(reason=reason)
        except Exception as exc:
            reason = f"{type(exc).__name__}: {exc}"
            _log.warning("council auto_spawn failed: %s", reason, exc_info=True)
            deliver_council_spawn_refusal(thread_id, reason=reason)
            out["early_stop"] = True
            out["spawn_failed"] = True
            out["message"] = ""
            out["prompt_tail"] = COUNCIL_SPAWN_FAILED_TAIL.format(reason=reason)
        return out

    return out
