"""Restart-safe automatic replies over Huihui's self-hosted BOXIM."""

import asyncio
import hashlib
import logging
import re
import secrets
import time
from datetime import datetime, timedelta
from typing import Callable

from sqlalchemy.orm import Session

from models import (
    Avatar,
    TakeoverCursor,
    TakeoverMessage,
    TakeoverReplyTask,
    User,
)
from services.boxim_client import BoxIMClient, BoxIMError

logger = logging.getLogger(__name__)

ACTIVE_TASK_STATUSES = ("pending", "generating", "ready", "sending")
GENERATABLE_TASK_STATUSES = ("pending",)
MAX_PROMPT_LENGTH = 4000
MAX_STALE_SECONDS = 120
STUCK_LOCK_SECONDS = 90
TAKEOVER_PERMISSION = "takeover"
TAKEOVER_DELAY_KEY = "takeoverReplyDelaySeconds"
DEFAULT_REPLY_DELAY_SECONDS = 180
MIN_REPLY_DELAY_SECONDS = 3
MAX_REPLY_DELAY_SECONDS = 86_400
HUMAN_PAUSE_SECONDS = 600
RATE_LIMIT_WINDOW_SECONDS = 300
RATE_LIMIT_MAX_REPLIES = 5
AVATAR_LOCAL_ID_PREFIX = "880"


def _utcnow() -> datetime:
    return datetime.utcnow()


def _takeover_enabled(avatar: Avatar | None) -> bool:
    if not avatar or avatar.status != "active":
        return False
    permissions = (avatar.config or {}).get("authorizationPermissions", [])
    return isinstance(permissions, list) and TAKEOVER_PERMISSION in permissions


def _boxim_time(value, fallback: datetime) -> datetime:
    try:
        timestamp = float(value)
        if timestamp > 10_000_000_000:
            timestamp /= 1000
        return datetime.utcfromtimestamp(timestamp)
    except (TypeError, ValueError, OSError, OverflowError):
        return fallback


def _numeric_id(value) -> int:
    try:
        return int(value)
    except (TypeError, ValueError):
        return 0


def _plain_text_reply(value: str) -> str:
    """BOXIM is plain text, so remove Markdown markers without damaging paragraphs."""
    text = (value or "").replace("\r\n", "\n").replace("\r", "\n")
    text = re.sub(r"```(?:\w+)?\n?(.*?)```", r"\1", text, flags=re.S)
    text = re.sub(r"\*\*(.*?)\*\*|__(.*?)__", lambda m: m.group(1) or m.group(2), text)
    text = re.sub(r"(?<!\*)\*([^*\n]+)\*(?!\*)", r"\1", text)
    text = re.sub(r"`([^`]+)`", r"\1", text)
    text = re.sub(r"^\s{0,3}#{1,6}\s*", "", text, flags=re.M)
    lines = [line.strip() for line in text.split("\n")]
    return "\n".join(line for line in lines if line).strip()


def _avatar_local_id(owner_id: str, trigger_message_id: str) -> str:
    """Build a deterministic BOXIM idempotency key that also marks avatar traffic."""
    digest = hashlib.sha256(f"{owner_id}:{trigger_message_id}".encode("utf-8")).digest()
    suffix = int.from_bytes(digest[:8], "big") % (10**15)
    return f"{AVATAR_LOCAL_ID_PREFIX}{suffix:015d}"


def _is_avatar_local_id(value: str | None) -> bool:
    local_id = str(value or "").strip()
    return len(local_id) == 18 and local_id.isdigit() and local_id.startswith(AVATAR_LOCAL_ID_PREFIX)


def _configured_reply_delay(avatar: Avatar, fallback: int | None = None) -> int:
    raw = (avatar.config or {}).get(
        TAKEOVER_DELAY_KEY,
        fallback if fallback is not None else DEFAULT_REPLY_DELAY_SECONDS,
    )
    if isinstance(raw, bool):
        return DEFAULT_REPLY_DELAY_SECONDS
    try:
        delay = int(raw)
    except (TypeError, ValueError):
        return DEFAULT_REPLY_DELAY_SECONDS
    if not MIN_REPLY_DELAY_SECONDS <= delay <= MAX_REPLY_DELAY_SECONDS:
        return DEFAULT_REPLY_DELAY_SECONDS
    return delay


class TakeoverService:
    """Poll BOXIM, honor the owner grace period, then generate and send one reply."""

    def __init__(
        self,
        session_factory: Callable[[], Session],
        boxim_client: BoxIMClient,
        *,
        reply_delay_seconds: int | None = None,
        now: Callable[[], datetime] = _utcnow,
    ):
        self.session_factory = session_factory
        self.boxim = boxim_client
        self.reply_delay_seconds = reply_delay_seconds
        self.now = now
        self._sessions: dict[str, dict] = {}
        self._poll_lock = asyncio.Lock()
        self._process_lock = asyncio.Lock()

    async def poll_and_process_messages(self):
        """Run one complete cycle for callers that do not use the split scheduler."""
        await self.poll_messages()
        await self.process_reply_tasks()

    async def poll_messages(self):
        """Fetch BOXIM events without blocking reply generation and dispatch."""
        if self._poll_lock.locked():
            return
        async with self._poll_lock:
            self._recover_stuck_tasks()
            avatar_ids = self._enabled_avatar_ids()
            self._cancel_disabled_tasks(set(avatar_ids))
            for avatar_id in avatar_ids:
                await self._sync_avatar(avatar_id)

    async def process_reply_tasks(self):
        """Generate and send replies independently from BOXIM's long poll."""
        if self._process_lock.locked():
            return
        async with self._process_lock:
            self._recover_stuck_tasks()
            avatar_ids = set(self._enabled_avatar_ids())
            self._cancel_disabled_tasks(avatar_ids)
            await self._prepare_replies()
            await self._dispatch_ready_replies()

    def _enabled_avatar_ids(self) -> list[str]:
        db = self.session_factory()
        try:
            avatars = (
                db.query(Avatar)
                .filter(Avatar.status == "active")
                .order_by(Avatar.updated_at.desc(), Avatar.created_at.desc())
                .all()
            )
            selected = {}
            for avatar in avatars:
                if _takeover_enabled(avatar) and avatar.owner_id not in selected:
                    selected[avatar.owner_id] = avatar.id
            return list(selected.values())
        finally:
            db.close()

    def _cancel_disabled_tasks(self, enabled_avatar_ids: set[str]):
        db = self.session_factory()
        try:
            tasks = (
                db.query(TakeoverReplyTask)
                .filter(TakeoverReplyTask.status.in_(ACTIVE_TASK_STATUSES))
                .all()
            )
            changed = False
            for task in tasks:
                if task.avatar_id not in enabled_avatar_ids:
                    task.status = "cancelled"
                    task.cancel_reason = "takeover_disabled"
                    task.locked_at = None
                    changed = True
            if changed:
                db.commit()
        finally:
            db.close()

    def _recover_stuck_tasks(self):
        db = self.session_factory()
        try:
            threshold = self.now() - timedelta(seconds=STUCK_LOCK_SECONDS)
            tasks = (
                db.query(TakeoverReplyTask)
                .filter(
                    TakeoverReplyTask.status.in_(("generating", "sending")),
                    TakeoverReplyTask.locked_at.isnot(None),
                    TakeoverReplyTask.locked_at < threshold,
                )
                .all()
            )
            for task in tasks:
                task.status = "pending" if task.status == "generating" else "ready"
                task.locked_at = None
                task.last_error = "上次处理意外中断，已自动恢复"
            if tasks:
                db.commit()
        finally:
            db.close()

    async def _boxim_session(self, user: User) -> dict:
        token_fingerprint = hashlib.sha256((user.huihui_token or "").encode()).hexdigest()
        cached = self._sessions.get(user.id)
        if (
            cached
            and cached["expires_at"] > time.monotonic()
            and cached["token_fingerprint"] == token_fingerprint
        ):
            return cached

        token_data = await self.boxim.exchange_access_token(user.huihui_token)
        access_token = token_data["accessToken"]
        profile = await self.boxim.get_self(access_token)
        try:
            expires_in = int(token_data.get("accessTokenExpiresIn") or 3600)
        except (TypeError, ValueError):
            expires_in = 3600
        if expires_in > 86_400:
            expires_in //= 1000
        cache_for = max(60, min(expires_in - 60, 3600))
        cached = {
            "access_token": access_token,
            "boxim_owner_id": str(profile["id"]),
            "expires_at": time.monotonic() + cache_for,
            "token_fingerprint": token_fingerprint,
        }
        self._sessions[user.id] = cached
        return cached

    def _forget_boxim_session(self, user_id: str):
        self._sessions.pop(user_id, None)

    def _record_connection_failure(
        self,
        db: Session,
        avatar: Avatar,
        cursor: TakeoverCursor,
        message: str,
        *,
        disable_takeover: bool,
    ):
        cursor.last_error = message
        cursor.last_polled_at = self.now()
        if not disable_takeover:
            return

        permissions = (avatar.config or {}).get("authorizationPermissions", [])
        avatar.config = {
            **(avatar.config or {}),
            "authorizationPermissions": [
                permission
                for permission in permissions
                if permission != TAKEOVER_PERMISSION
            ],
        }
        tasks = (
            db.query(TakeoverReplyTask)
            .filter(
                TakeoverReplyTask.avatar_id == avatar.id,
                TakeoverReplyTask.status.in_(ACTIVE_TASK_STATUSES),
            )
            .all()
        )
        for task in tasks:
            task.status = "cancelled"
            task.cancel_reason = "connection_failed"
            task.locked_at = None

    async def _sync_avatar(self, avatar_id: str) -> bool:
        db = self.session_factory()
        try:
            avatar = db.query(Avatar).filter(Avatar.id == avatar_id).first()
            if not _takeover_enabled(avatar):
                return False
            user = db.query(User).filter(User.huihui_user_id == avatar.owner_id).first()
            cursor = db.query(TakeoverCursor).filter(TakeoverCursor.avatar_id == avatar.id).first()
            if not cursor:
                cursor = TakeoverCursor(avatar_id=avatar.id, owner_id=avatar.owner_id)
                db.add(cursor)
                db.flush()
            if not user or not user.huihui_token:
                self._record_connection_failure(
                    db,
                    avatar,
                    cursor,
                    "请重新登录会会生产账号后再开启主动接管",
                    disable_takeover=True,
                )
                db.commit()
                return False

            try:
                session = await self._boxim_session(user)
                owner_boxim_id = session["boxim_owner_id"]
                if cursor.boxim_owner_id and cursor.boxim_owner_id != owner_boxim_id:
                    cursor.initialized = False
                    cursor.last_message_id = "0"
                cursor.boxim_owner_id = owner_boxim_id
                messages = await self.boxim.fetch_private_messages(
                    session["access_token"], cursor.last_message_id or "0"
                )
            except Exception as exc:
                if isinstance(exc, BoxIMError) and exc.auth_error:
                    self._forget_boxim_session(user.id)
                    message = "BOXIM 授权已失效，请重新登录会会生产账号"
                    disable_takeover = True
                else:
                    message = f"BOXIM 暂时连接失败：{str(exc)[:160]}"
                    disable_takeover = False
                self._record_connection_failure(
                    db,
                    avatar,
                    cursor,
                    message,
                    disable_takeover=disable_takeover,
                )
                db.commit()
                logger.warning(
                    "BOXIM sync failed for avatar %s (will_retry=%s): %s",
                    avatar.id,
                    not disable_takeover,
                    exc,
                )
                return False

            messages.sort(key=lambda item: (_numeric_id(item.get("id")), item.get("sendTime") or 0))
            priming = not bool(cursor.initialized)
            max_message_id = _numeric_id(cursor.last_message_id)
            read_receipts: dict[str, int] = {}
            for message in messages:
                self._record_message(
                    db,
                    avatar,
                    cursor.boxim_owner_id,
                    message,
                    schedule_reply=not priming,
                )
                message_id = _numeric_id(message.get("id"))
                max_message_id = max(max_message_id, message_id)
                send_id = str(message.get("sendId") or "")
                recv_id = str(message.get("recvId") or "")
                if recv_id == cursor.boxim_owner_id and send_id and message_id:
                    read_receipts[send_id] = max(read_receipts.get(send_id, 0), message_id)

            # BOXIM publishes this HTTP state change to connected socket clients.
            # Do it before advancing the cursor so a failed receipt is retried.
            for peer_id, message_id in read_receipts.items():
                await self.boxim.mark_private_messages_read(
                    session["access_token"], peer_id, message_id
                )

            cursor.last_message_id = str(max_message_id)
            cursor.initialized = True
            cursor.last_polled_at = self.now()
            cursor.last_error = ""
            db.commit()
            return True
        except Exception:
            db.rollback()
            logger.exception("Failed to persist BOXIM messages for avatar %s", avatar_id)
            return False
        finally:
            db.close()

    def _record_message(
        self,
        db: Session,
        avatar: Avatar,
        boxim_owner_id: str,
        message: dict,
        *,
        schedule_reply: bool,
    ):
        message_id = str(message.get("id") or "").strip()
        if not message_id:
            return
        local_id = str(message.get("localId") or "").strip() or None
        if (
            db.query(TakeoverMessage)
            .filter(
                TakeoverMessage.owner_id == avatar.owner_id,
                TakeoverMessage.boxim_message_id == message_id,
            )
            .first()
        ):
            return

        send_id = str(message.get("sendId") or "")
        recv_id = str(message.get("recvId") or "")
        if send_id == boxim_owner_id:
            direction, peer_id = "outgoing", recv_id
        elif recv_id == boxim_owner_id:
            direction, peer_id = "incoming", send_id
        else:
            return
        if not peer_id:
            return

        now = self.now()
        send_time = _boxim_time(message.get("sendTime"), now)
        is_avatar = _is_avatar_local_id(local_id)
        if not is_avatar and local_id:
            is_avatar = bool(
                db.query(TakeoverReplyTask)
                .filter(
                    TakeoverReplyTask.boxim_local_id == local_id,
                    TakeoverReplyTask.status.in_(("ready", "sending", "sent")),
                )
                .first()
            )
        if not is_avatar:
            is_avatar = bool(
                db.query(TakeoverReplyTask)
                .filter(
                    TakeoverReplyTask.boxim_sent_message_id == message_id,
                    TakeoverReplyTask.status == "sent",
                )
                .first()
            )

        event = TakeoverMessage(
            avatar_id=avatar.id,
            owner_id=avatar.owner_id,
            boxim_message_id=message_id,
            boxim_local_id=local_id,
            peer_id=peer_id,
            direction=direction,
            message_type=int(message.get("type") or 0),
            content=str(message.get("content") or ""),
            is_avatar=is_avatar,
            send_time=send_time,
        )
        db.add(event)
        db.flush()

        if direction == "outgoing":
            if not is_avatar:
                self._cancel_conversation(db, avatar.owner_id, peer_id, "owner_replied")
            return
        if not schedule_reply or event.message_type != 0 or not event.content.strip():
            return
        if (now - send_time).total_seconds() > MAX_STALE_SECONDS:
            return
        if is_avatar:
            self._cancel_conversation(db, avatar.owner_id, peer_id, "peer_avatar_message")
            return
        if self._human_pause_active(db, avatar.owner_id, peer_id, now):
            self._cancel_conversation(db, avatar.owner_id, peer_id, "owner_active")
            return
        if self._conversation_rate_limited(db, avatar.owner_id, peer_id, now):
            self._cancel_conversation(db, avatar.owner_id, peer_id, "rate_limited")
            return
        self._schedule_reply(db, avatar, event)

    @staticmethod
    def _human_pause_active(db: Session, owner_id: str, peer_id: str, now: datetime) -> bool:
        threshold = now - timedelta(seconds=HUMAN_PAUSE_SECONDS)
        return bool(
            db.query(TakeoverMessage.id)
            .filter(
                TakeoverMessage.owner_id == owner_id,
                TakeoverMessage.peer_id == peer_id,
                TakeoverMessage.direction == "outgoing",
                TakeoverMessage.is_avatar.is_(False),
                TakeoverMessage.send_time >= threshold,
            )
            .first()
        )

    @staticmethod
    def _conversation_rate_limited(
        db: Session,
        owner_id: str,
        peer_id: str,
        now: datetime,
    ) -> bool:
        threshold = now - timedelta(seconds=RATE_LIMIT_WINDOW_SECONDS)
        return (
            db.query(TakeoverReplyTask.id)
            .filter(
                TakeoverReplyTask.owner_id == owner_id,
                TakeoverReplyTask.peer_id == peer_id,
                TakeoverReplyTask.status == "sent",
                TakeoverReplyTask.sent_at >= threshold,
            )
            .count()
            >= RATE_LIMIT_MAX_REPLIES
        )

    @staticmethod
    def _cancel_conversation(db: Session, owner_id: str, peer_id: str, reason: str):
        tasks = (
            db.query(TakeoverReplyTask)
            .filter(
                TakeoverReplyTask.owner_id == owner_id,
                TakeoverReplyTask.peer_id == peer_id,
                TakeoverReplyTask.status.in_(ACTIVE_TASK_STATUSES),
            )
            .all()
        )
        for task in tasks:
            task.status = "cancelled"
            task.cancel_reason = reason
            task.locked_at = None

    def _schedule_reply(self, db: Session, avatar: Avatar, event: TakeoverMessage):
        active_tasks = (
            db.query(TakeoverReplyTask)
            .filter(
                TakeoverReplyTask.owner_id == avatar.owner_id,
                TakeoverReplyTask.peer_id == event.peer_id,
                TakeoverReplyTask.status.in_(("pending", "generating", "ready")),
            )
            .order_by(TakeoverReplyTask.created_at.desc())
            .all()
        )
        prompt_parts = []
        source_ids = []
        if active_tasks:
            latest = active_tasks[0]
            prompt_parts.append(latest.prompt)
            source_ids.extend(latest.source_message_ids or [])
            for task in active_tasks:
                task.status = "cancelled"
                task.cancel_reason = "newer_incoming_message"
                task.locked_at = None
        prompt_parts.append(event.content.strip())
        source_ids.append(event.boxim_message_id)
        prompt = "\n".join(part for part in prompt_parts if part).strip()[-MAX_PROMPT_LENGTH:]
        due_at = event.send_time + timedelta(
            seconds=_configured_reply_delay(avatar, self.reply_delay_seconds)
        )
        task_id = secrets.token_hex(16)
        local_id = _avatar_local_id(avatar.owner_id, event.boxim_message_id)
        db.add(
            TakeoverReplyTask(
                id=task_id,
                avatar_id=avatar.id,
                owner_id=avatar.owner_id,
                peer_id=event.peer_id,
                trigger_message_id=event.boxim_message_id,
                source_message_ids=source_ids,
                prompt=prompt,
                status="pending",
                scheduled_at=due_at,
                boxim_local_id=str(local_id),
            )
        )

    async def _prepare_replies(self) -> int:
        db = self.session_factory()
        try:
            task_ids = [
                row[0]
                for row in (
                    db.query(TakeoverReplyTask.id)
                    .filter(
                        TakeoverReplyTask.status.in_(GENERATABLE_TASK_STATUSES),
                        TakeoverReplyTask.response_text == "",
                        TakeoverReplyTask.scheduled_at <= self.now(),
                    )
                    .order_by(TakeoverReplyTask.created_at.asc())
                    .limit(10)
                    .all()
                )
            ]
        finally:
            db.close()

        if not task_ids:
            return 0

        # Each conversation owns its task, so unrelated contacts can generate in
        # parallel instead of one slow model response delaying every other peer.
        semaphore = asyncio.Semaphore(4)

        async def generate(task_id: str) -> bool:
            async with semaphore:
                return await asyncio.to_thread(self._generate_reply, task_id)

        results = await asyncio.gather(*(generate(task_id) for task_id in task_ids))
        return sum(bool(result) for result in results)

    def _generate_reply(self, task_id: str) -> bool:
        db = self.session_factory()
        try:
            task = db.query(TakeoverReplyTask).filter(TakeoverReplyTask.id == task_id).first()
            if not task or task.status != "pending":
                return False
            avatar = db.query(Avatar).filter(Avatar.id == task.avatar_id).first()
            if not _takeover_enabled(avatar):
                task.status = "cancelled"
                task.cancel_reason = "takeover_disabled"
                db.commit()
                return False

            task.status = "generating"
            task.locked_at = self.now()
            db.commit()

            excluded_ids = set(task.source_message_ids or [])
            events = (
                db.query(TakeoverMessage)
                .filter(
                    TakeoverMessage.owner_id == task.owner_id,
                    TakeoverMessage.peer_id == task.peer_id,
                    TakeoverMessage.avatar_id == task.avatar_id,
                )
                .order_by(TakeoverMessage.send_time.desc())
                .limit(30)
                .all()
            )
            history = []
            for event in reversed(events):
                if event.boxim_message_id in excluded_ids or not event.content.strip():
                    continue
                if event.direction == "incoming" and event.is_avatar:
                    continue
                history.append(
                    {
                        "role": "user" if event.direction == "incoming" else "assistant",
                        "content": event.content.strip(),
                    }
                )
            history = history[-10:]

            from routers.chat import _resolve_reply

            result = _resolve_reply(db, avatar, task.prompt, history, usage_source="takeover")
            answer = _plain_text_reply(result.get("answer", ""))
            db.refresh(task)
            if task.status != "generating":
                return False
            if not answer:
                raise RuntimeError("分身没有生成有效回复")
            task.response_text = answer
            task.status = "ready"
            task.locked_at = None
            task.last_error = ""
            db.commit()
            return True
        except Exception as exc:
            db.rollback()
            task = db.query(TakeoverReplyTask).filter(TakeoverReplyTask.id == task_id).first()
            if task and task.status in ("pending", "generating"):
                task.attempts = (task.attempts or 0) + 1
                task.status = "pending" if task.attempts < 3 else "failed"
                task.locked_at = None
                task.last_error = str(exc)[:300]
                db.commit()
            logger.warning("Failed to prepare takeover reply %s: %s", task_id, exc)
            return False
        finally:
            db.close()

    async def _dispatch_ready_replies(self):
        db = self.session_factory()
        try:
            task_ids = [
                row[0]
                for row in (
                    db.query(TakeoverReplyTask.id)
                    .filter(
                        TakeoverReplyTask.status == "ready",
                        TakeoverReplyTask.scheduled_at <= self.now(),
                    )
                    .order_by(TakeoverReplyTask.scheduled_at.asc())
                    .limit(10)
                    .all()
                )
            ]
        finally:
            db.close()

        if task_ids:
            await asyncio.gather(*(self._send_task(task_id) for task_id in task_ids))

    async def _send_task(self, task_id: str) -> bool:
        db = self.session_factory()
        user = None
        try:
            task = db.query(TakeoverReplyTask).filter(TakeoverReplyTask.id == task_id).first()
            if not task or task.status != "ready":
                return False
            avatar = db.query(Avatar).filter(Avatar.id == task.avatar_id).first()
            if not _takeover_enabled(avatar):
                task.status = "cancelled"
                task.cancel_reason = "takeover_disabled"
                db.commit()
                return False
            if (self.now() - task.scheduled_at).total_seconds() > MAX_STALE_SECONDS:
                task.status = "cancelled"
                task.cancel_reason = "stale_reply"
                db.commit()
                return False
            cursor = (
                db.query(TakeoverCursor)
                .filter(TakeoverCursor.avatar_id == task.avatar_id)
                .first()
            )
            if not cursor or not cursor.last_polled_at or cursor.last_polled_at < task.scheduled_at:
                # Do not race the owner's final seconds of the grace period. A
                # completed poll at/after the due time must confirm no human reply.
                return False
            user = db.query(User).filter(User.huihui_user_id == task.owner_id).first()
            if not user or not user.huihui_token:
                raise BoxIMError("缺少会会登录凭证", auth_error=True)

            task.status = "sending"
            task.locked_at = self.now()
            db.commit()
            session = await self._boxim_session(user)
            result = await self.boxim.send_private_message(
                session["access_token"],
                task.peer_id,
                task.response_text,
                local_id=task.boxim_local_id,
            )
            db.refresh(task)
            if task.status != "sending":
                return False
            task.status = "sent"
            task.sent_at = self.now()
            task.locked_at = None
            task.last_error = ""
            task.boxim_sent_message_id = str(result.get("id") or "")
            db.commit()
            logger.info("BOXIM takeover reply sent for task %s", task.id)
            return True
        except Exception as exc:
            db.rollback()
            if user and isinstance(exc, BoxIMError) and exc.auth_error:
                self._forget_boxim_session(user.id)
            task = db.query(TakeoverReplyTask).filter(TakeoverReplyTask.id == task_id).first()
            if task and task.status in ("ready", "sending"):
                task.attempts = (task.attempts or 0) + 1
                task.status = "ready" if task.attempts < 3 else "failed"
                task.locked_at = None
                task.last_error = str(exc)[:300]
                if task.status == "ready":
                    task.scheduled_at = self.now() + timedelta(seconds=2 ** task.attempts)
                db.commit()
            logger.warning("Failed to send takeover reply %s: %s", task_id, exc)
            return False
        finally:
            db.close()