"""Cookie 驱动的 Gemini 会话生命周期管理。

本模块把「基于 Cookie 初始化客户端」和「进程常驻期间持续自动轮换
``__Secure-1PSIDTS``」这两件事收敛到一个地方，供 ``main.py`` 直接使用。

为什么要单独写这一层，而不是直接 ``GeminiClient.init(auto_refresh=True)``：

1. **统一凭据来源。** ``SECURE_1PSID`` / ``SECURE_1PSIDTS`` 环境变量、上一次轮换
   落盘的文本缓存、gemini-webapi 自己的 JSON 缓存，三者需要按「谁更新用谁」的
   顺序收敛成一个候选序列，逐个尝试初始化。
2. **补上缺失的自愈能力。** gemini-webapi 的后台轮换任务在账号状态不是
   ``AVAILABLE`` 时会直接 ``break`` 并把 ``refresh_task`` 置空，活跃度看门狗同理；
   之后再没有任何机制把它拉起来。对 7x24 常驻的 HF Space 来说，只要出现一次
   「配额耗尽 / 临时不可用」，Cookie 就再也不会被轮换，直到进程重启——这正是
   「进程一直在跑但 1PSIDTS 还是过期了」的根因。本模块用监督协程负责重新拉起。
3. **可观测。** 轮换是否发生、最后一次换值距今多久、后台任务是否还活着，都能通过
   ``status()`` 暴露给管理面板。
"""

from __future__ import annotations

import asyncio
import json
import logging
import os
import re
import time
from pathlib import Path
from typing import Any, Awaitable, Callable, Optional

from gemini_webapi import GeminiClient
from gemini_webapi.client import rotate_1psidts

logger = logging.getLogger(__name__)

PSID_COOKIE = "__Secure-1PSID"
PSIDTS_COOKIE = "__Secure-1PSIDTS"

_SAFE_PSID_PATTERN = re.compile(r"^[\w\-.]+$")

CredentialProvider = Callable[[], tuple[str, str]]
Validator = Callable[[GeminiClient, str], Awaitable[None]]


def is_safe_psid(psid: str) -> bool:
	"""判断 1PSID 能否安全地拼进缓存文件名。"""
	return bool(psid) and bool(_SAFE_PSID_PATTERN.match(psid))


def extract_cookie(cookies: Any, name: str) -> str:
	"""从 curl_cffi / httpx 的 cookie 容器中读取指定 cookie 的值。

	直接遍历 ``cookies.jar``，这样对 httpx 那种「必须带 domain 才能 get 到」的
	容器同样有效。
	"""
	if cookies is None:
		return ""

	jar = getattr(cookies, "jar", None)
	if jar is not None:
		try:
			for cookie in jar:
				if getattr(cookie, "name", None) == name:
					return getattr(cookie, "value", "") or ""
		except TypeError:
			pass

	try:
		return cookies.get(name) or ""
	except Exception:
		return ""


def mask_secret(value: str, keep: int = 4) -> str:
	"""脱敏显示，用于日志和管理面板。"""
	if not value:
		return ""
	if len(value) <= keep * 2:
		return "*" * len(value)
	return f"{value[:keep]}{'*' * 6}{value[-keep:]}"


class CookieStore:
	"""统一管理 ``__Secure-1PSIDTS`` 的落盘缓存。

	同时维护两份文件，读取时取「修改时间更新」的那一份：

	* ``.cached_1psidts_{psid}.txt`` —— 纯文本，只存最新值。便于人工查看和复制，
	  也是排查凭据问题最直观的来源。
	* ``.cached_cookies_{psid}.json`` —— gemini-webapi 自带格式，``get_access_token``
	  初始化时会优先读取它，所以必须让两边保持一致，否则会出现「缓存里是旧值、
	  但初始化却用了它」的诡异情况。
	"""

	def __init__(self, cache_dir: str | os.PathLike[str]):
		self.cache_dir = Path(cache_dir)

	def text_path(self, psid: str) -> Optional[Path]:
		if not is_safe_psid(psid):
			return None
		return self.cache_dir / f".cached_1psidts_{psid}.txt"

	def json_path(self, psid: str) -> Optional[Path]:
		if not psid:
			return None
		return self.cache_dir / f".cached_cookies_{psid}.json"

	def read_1psidts(self, psid: str) -> tuple[str, str]:
		"""返回 ``(值, 来源)``，来源取值 ``text-cache`` / ``json-cache`` / ``""``。"""
		candidates: list[tuple[float, str, str]] = []

		text_path = self.text_path(psid)
		if text_path is not None and text_path.is_file():
			try:
				value = text_path.read_text(encoding="utf-8").strip()
				if value:
					candidates.append((text_path.stat().st_mtime, value, "text-cache"))
			except OSError as e:
				logger.warning("Failed to read 1PSIDTS text cache %s: %s", text_path, e)

		json_path = self.json_path(psid)
		if json_path is not None and json_path.is_file():
			try:
				value = self._read_from_json(json_path)
				if value:
					candidates.append((json_path.stat().st_mtime, value, "json-cache"))
			except (OSError, ValueError) as e:
				logger.warning("Failed to read cookie cache %s: %s", json_path, e)

		if not candidates:
			return "", ""

		candidates.sort(key=lambda item: item[0], reverse=True)
		_, value, source = candidates[0]
		return value, source

	def write_1psidts(self, psid: str, value: str) -> bool:
		"""把最新 1PSIDTS 写入文本缓存（原子替换）。返回是否真的写入了新值。"""
		if not value:
			return False

		path = self.text_path(psid)
		if path is None:
			logger.warning("Skip caching 1PSIDTS: __Secure-1PSID is not a safe filename token.")
			return False

		try:
			existing = path.read_text(encoding="utf-8").strip() if path.is_file() else ""
		except OSError:
			existing = ""
		if existing == value:
			return False

		try:
			path.parent.mkdir(parents=True, exist_ok=True)
			tmp_path = path.parent / f"{path.name}.{os.getpid()}.tmp"
			tmp_path.write_text(value, encoding="utf-8")
			os.replace(tmp_path, path)
			try:
				os.chmod(path, 0o600)
			except OSError:
				pass
			return True
		except OSError as e:
			logger.warning("Failed to persist 1PSIDTS to %s: %s", path, e)
			return False

	@staticmethod
	def _read_from_json(path: Path) -> str:
		payload = json.loads(path.read_text(encoding="utf-8"))
		if isinstance(payload, dict):
			payload = payload.get("cookies", [])
		for cookie in payload or []:
			if isinstance(cookie, dict) and cookie.get("name") == PSIDTS_COOKIE:
				return cookie.get("value", "") or ""
		return ""


class HfCookieBackup:
	"""把 Cookie 缓存备份到 Hugging Face，用于跨重启续期（可选）。

	两种目标可以单独或同时开启：

	* ``dataset_repo``：上传/下载缓存文件到私有 Dataset。**不会**触发 Space 重启，
	  推荐用这种方式做持久化。
	* ``space_repo``：回写 Space 的 Secret（默认 key 为 ``SECURE_1PSIDTS``）。HF 的
	  规则是「任何 Space 配置变更都会重启 Space」，所以这条路径默认关闭，并且带
	  最小间隔保护——它只在「进程已经要重启」这类场景下才有意义。
	"""

	def __init__(
		self,
		cache_dir: str | os.PathLike[str],
		token: str = "",
		dataset_repo: str = "",
		space_repo: str = "",
		secret_key: str = PSIDTS_COOKIE,
		min_interval: float = 3600,
	):
		self.store = CookieStore(cache_dir)
		self.token = token
		self.dataset_repo = dataset_repo
		self.space_repo = space_repo
		self.secret_key = secret_key
		self.min_interval = max(0.0, min_interval)
		self._last_push_at = 0.0
		self._api: Any = None

	@property
	def enabled(self) -> bool:
		return bool(self.token and (self.dataset_repo or self.space_repo))

	async def restore(self, psid: str) -> None:
		"""本地没有缓存时，尝试从 Dataset 拉回上次备份。"""
		if not (self.token and self.dataset_repo):
			return
		text_path = self.store.text_path(psid)
		if text_path is not None and text_path.is_file():
			return
		await asyncio.to_thread(self._restore_blocking, psid)

	async def push(self, psid: str, value: str, force: bool = False) -> dict:
		"""备份当前 1PSIDTS，带最小间隔保护。"""
		if not self.enabled:
			return {"skipped": "backup not configured"}
		now = time.time()
		if not force and now - self._last_push_at < self.min_interval:
			return {"skipped": "rate limited"}
		self._last_push_at = now
		return await asyncio.to_thread(self._push_blocking, psid, value)

	def _restore_blocking(self, psid: str) -> None:
		try:
			from huggingface_hub import hf_hub_download
		except ImportError:
			logger.warning("huggingface_hub is not installed; cookie backup is disabled.")
			return

		for path in (self.store.json_path(psid), self.store.text_path(psid)):
			if path is None or path.is_file():
				continue
			try:
				downloaded = hf_hub_download(
					repo_id=self.dataset_repo,
					filename=path.name,
					repo_type="dataset",
					token=self.token,
					local_dir=str(self.store.cache_dir),
				)
				logger.info("Restored cookie cache from HF dataset: %s", downloaded)
			except Exception as e:
				logger.debug("No remote cookie cache for %s: %s", path.name, e)

	def _push_blocking(self, psid: str, value: str) -> dict:
		try:
			from huggingface_hub import HfApi
		except ImportError:
			return {"skipped": "huggingface_hub is not installed"}

		result: dict = {}
		api = self._api or HfApi(token=self.token)
		self._api = api

		if self.dataset_repo:
			uploaded = []
			for path in (self.store.text_path(psid), self.store.json_path(psid)):
				if path is None or not path.is_file():
					continue
				try:
					api.upload_file(
						path_or_fileobj=str(path),
						path_in_repo=path.name,
						repo_id=self.dataset_repo,
						repo_type="dataset",
					)
					uploaded.append(path.name)
				except Exception as e:
					logger.warning("Failed to upload %s to HF dataset: %s", path.name, e)
			result["dataset_uploaded"] = uploaded

		if self.space_repo:
			try:
				api.add_space_secret(
					repo_id=self.space_repo,
					key=self.secret_key,
					value=value,
				)
				result["space_secret_updated"] = True
				logger.warning(
					"Wrote %s back to Space secret %s; note that HF restarts the Space on any config change.",
					self.secret_key,
					self.space_repo,
				)
			except Exception as e:
				logger.warning("Failed to update Space secret: %s", e)
				result["space_secret_updated"] = False

		return result


class GeminiSessionManager:
	"""维护「基于 Cookie 初始化」+「后台持续轮换 1PSIDTS」的会话生命周期。

	与直接 ``GeminiClient.init(auto_refresh=True)`` 的区别在于自愈能力：监督协程会
	周期性地检查会话健康度，发现后台轮换任务停止就重新拉起，发现会话失效就重建。
	"""

	def __init__(
		self,
		credential_provider: CredentialProvider,
		cache_dir: str | os.PathLike[str],
		validate: Optional[Validator] = None,
		backup: Optional[HfCookieBackup] = None,
		proxy: str = "",
		timeout: float = 300,
		watchdog_timeout: float = 120,
		refresh_interval: float = 600,
		health_interval: float = 120,
		rearm_cooldown: float = 300,
		init_retry_interval: float = 300,
		verbose: bool = False,
	):
		self.credential_provider = credential_provider
		self.validate = validate
		self.backup = backup
		self.proxy = proxy
		self.timeout = max(30.0, float(timeout))
		self.watchdog_timeout = max(30.0, float(watchdog_timeout))
		self.refresh_interval = max(60.0, float(refresh_interval))
		self.health_interval = max(30.0, float(health_interval))
		self.rearm_cooldown = max(60.0, float(rearm_cooldown))
		self.init_retry_interval = max(60.0, float(init_retry_interval))
		self.verbose = verbose

		self.client: Optional[GeminiClient] = None
		self.lock = asyncio.Lock()
		self.store = CookieStore(cache_dir)

		self._supervisor_task: Optional[asyncio.Task] = None
		self._rotation_task: Optional[asyncio.Task] = None
		self._closing = False
		self._next_init_at = 0.0
		self._rearm_until = 0.0

		self._source = ""
		self._last_error = ""
		self._last_1psidts = ""
		self._last_psidts_change_at = 0.0

		self._init_count = 0
		self._recycle_count = 0
		self._rearm_count = 0
		self._rotation_count = 0

	# ------------------------------------------------------------------ 生命周期

	async def start(self) -> None:
		"""启动后台监督协程（幂等）。"""
		if self._supervisor_task is not None and not self._supervisor_task.done():
			return
		self._closing = False
		self._supervisor_task = asyncio.create_task(self._supervisor(), name="gemini-session-supervisor")
		logger.info(
			"Gemini session supervisor started (health check every %ss, rotation interval %ss)",
			int(self.health_interval),
			int(self.refresh_interval),
		)

	async def stop(self) -> None:
		"""停止监督协程并关闭客户端（关闭时会把最新 Cookie 落盘）。"""
		self._closing = True
		task, self._supervisor_task = self._supervisor_task, None
		if task is not None and not task.done():
			task.cancel()
			try:
				await task
			except asyncio.CancelledError:
				pass
		rotation_task, self._rotation_task = self._rotation_task, None
		if rotation_task is not None and not rotation_task.done():
			rotation_task.cancel()
			try:
				await rotation_task
			except asyncio.CancelledError:
				pass
		async with self.lock:
			await self._drop_client_locked("service shutdown")

	# ------------------------------------------------------------------ 对外接口

	async def get_client(self) -> GeminiClient:
		"""返回一个可用的客户端；只在传输层失效时按需重建。

		``gemini-webapi`` 的用户状态探针可能因为 Google 的辅助 RPC、出口限制或
		短暂响应异常把 ``account_status`` 标为 ``UNAUTHENTICATED``，但主生成流仍然
		可以工作。这里不能仅凭这个字段关闭正在服务请求的客户端；真正的认证失败由
		生成请求的异常路径触发重建。
		"""
		client = self.client
		if client is not None:
			reason = self.session_transport_problem(client)
			if not reason:
				return client
			logger.warning("Gemini client transport is unusable (%s); rebuilding on demand", reason)

		async with self.lock:
			if self.client is not None:
				reason = self.session_transport_problem(self.client)
				if reason:
					await self._drop_client_locked(f"unusable transport: {reason}")
			if self.client is None:
				await self._initialize_locked()
			if self.client is None:
				raise RuntimeError(self._last_error or "Gemini client is unavailable")
			return self.client

	async def recover(self, reason: str = "manual recovery") -> GeminiClient:
		"""强制丢弃当前会话并重建，供请求失败重试 / 管理面板使用。"""
		async with self.lock:
			await self._drop_client_locked(reason)
			await self._initialize_locked()
			if self.client is None:
				raise RuntimeError(self._last_error or "Gemini client is unavailable")
			return self.client

	def status(self) -> dict:
		"""运行时状态快照，用于日志和管理面板展示。"""
		client = self.client
		psid, env_psidts = self.credential_provider()
		now = time.time()
		cached_psidts, cached_source = self.store.read_1psidts(psid)

		return {
			"initialized": client is not None,
			"supervisor_alive": bool(self._supervisor_task is not None and not self._supervisor_task.done()),
			"credential_source": self._source,
			"secure_1psid_masked": mask_secret(psid),
			"env_1psidts_masked": mask_secret(env_psidts),
			"live_1psidts_masked": mask_secret(self._last_1psidts),
			"cached_1psidts_masked": mask_secret(cached_psidts),
			"cached_1psidts_source": cached_source,
			"rotation_count": self._rotation_count,
			"last_1psidts_change_age_sec": int(now - self._last_psidts_change_at) if self._last_psidts_change_at else None,
			"rotation_task_alive": bool(
				client is not None and (
					(client.refresh_task is not None and not client.refresh_task.done()) or
					(self._rotation_task is not None and not self._rotation_task.done())
				)
			),
			"managed_rotation_task_alive": bool(self._rotation_task is not None and not self._rotation_task.done()),
			"auto_refresh": bool(getattr(client, "auto_refresh", False)),
			"refresh_interval": int(self.refresh_interval),
			"request_timeout": int(self.timeout),
			"watchdog_timeout": int(self.watchdog_timeout),
			"rearm_count": self._rearm_count,
			"recycle_count": self._recycle_count,
			"init_count": self._init_count,
			"account_status": getattr(getattr(client, "account_status", None), "name", "unknown"),
			"last_error": self._last_error,
			"cache_dir": str(self.store.cache_dir),
			"backup_enabled": bool(self.backup is not None and self.backup.enabled),
		}

	# ------------------------------------------------------------------ 内部实现

	async def _initialize_locked(self) -> None:
		"""按「缓存 -> 环境变量」的顺序逐个尝试初始化。调用方需持有 self.lock。"""
		psid, env_psidts = self.credential_provider()
		if not psid:
			self._last_error = "SECURE_1PSID is not configured"
			logger.error("%s", self._last_error)
			return

		cached_psidts, cached_source = self.store.read_1psidts(psid)

		if self.backup is not None and not cached_psidts:
			await self.backup.restore(psid)
			cached_psidts, cached_source = self.store.read_1psidts(psid)

		attempts: list[tuple[str, str]] = []
		seen: set[str] = set()
		for source, psidts in (
			(f"cache ({cached_source})" if cached_source else "cache", cached_psidts),
			("environment", env_psidts),
		):
			if not psidts or psidts in seen:
				continue
			seen.add(psidts)
			attempts.append((source, psidts))

		if not attempts:
			self._last_error = "SECURE_1PSIDTS is missing and no rotated value has been cached yet"
			logger.error("%s", self._last_error)
			return

		self._init_count += 1
		last_error: Optional[BaseException] = None

		for source, psidts in attempts:
			kwargs = {"proxy": self.proxy} if self.proxy else {}
			candidate = GeminiClient(psid, psidts, **kwargs)
			try:
				logger.info("Initializing Gemini client using %s credentials", source)
				await candidate.init(
					timeout=self.timeout,
					auto_refresh=True,
					refresh_interval=self.refresh_interval,
					auto_close=False,
					watchdog_timeout=self.watchdog_timeout,
					verbose=self.verbose,
				)
				if self.validate is not None:
					await self.validate(candidate, source)
			except Exception as e:
				last_error = e
				logger.warning("Gemini session setup failed using %s 1PSIDTS: %s", source, e)
				try:
					await candidate.close()
				except Exception:
					pass
				continue

			self.client = candidate
			if self._rotation_task is not None and not self._rotation_task.done():
				self._rotation_task.cancel()
			self._rotation_task = asyncio.create_task(self._rotation_supervisor(candidate), name="gemini-cookie-rotation-supervisor")
			self._source = source
			self._last_error = ""
			live_psidts = extract_cookie(candidate.cookies, PSIDTS_COOKIE) or psidts
			self._last_1psidts = live_psidts
			self._last_psidts_change_at = time.time()
			if self.store.write_1psidts(psid, live_psidts):
				logger.info("Seeded __Secure-1PSIDTS cache from %s credentials", source)
			logger.info(
				"Gemini client ready (source=%s, 1psidts=%s, auto_refresh=%s, interval=%ss)",
				source,
				mask_secret(live_psidts),
				candidate.auto_refresh,
				int(self.refresh_interval),
			)
			return

		self._last_error = str(last_error) if last_error is not None else "initialization failed"
		logger.error("Failed to initialize Gemini client: %s", self._last_error)

	async def _drop_client_locked(self, reason: str) -> None:
		"""关闭并丢弃当前客户端。调用方需持有 self.lock。"""
		client, self.client = self.client, None
		if client is None:
			return

		if self._rotation_task is not None and not self._rotation_task.done():
			self._rotation_task.cancel()
		self._rotation_task = None

		self._recycle_count += 1
		logger.warning("Dropping Gemini client (%s)", reason)

		# 关闭前先把当前 1PSIDTS 落盘；close() 内部也会保存 gemini-webapi 的 JSON 缓存
		psid, _ = self.credential_provider()
		live_psidts = extract_cookie(getattr(client, "cookies", None), PSIDTS_COOKIE)
		if live_psidts:
			self._last_1psidts = live_psidts
			self.store.write_1psidts(psid, live_psidts)

		try:
			await client.close()
		except Exception as e:
			logger.warning("Failed to close Gemini client: %s", e)

	@staticmethod
	def session_transport_problem(client: GeminiClient) -> str:
		"""判断客户端传输层是否已关闭，不采信易误报的账号状态探针。"""
		if client is None:
			return "client is missing"
		if not getattr(client, "_running", False):
			return "client is not running"
		if getattr(client, "client", None) is None:
			return "underlying HTTP session is closed"
		return ""

	@classmethod
	def session_problem(cls, client: GeminiClient) -> str:
		"""返回完整诊断；认证状态供异常恢复判断，不再触发后台主动关闭。"""
		reason = cls.session_transport_problem(client)
		if reason:
			return reason
		account_status = getattr(client, "account_status", None)
		if getattr(account_status, "name", "") == "UNAUTHENTICATED":
			return "account status is UNAUTHENTICATED"
		return ""

	async def _rotation_supervisor(self, client: GeminiClient) -> None:
		"""轮换 Cookie，不依赖 gemini-webapi 的辅助账号状态探针。"""
		while not self._closing and self.client is client and getattr(client, "_running", False):
			try:
				await asyncio.sleep(max(60.0, self.refresh_interval))
			except asyncio.CancelledError:
				raise
			if self._closing or self.client is not client or not getattr(client, "_running", False):
				return
			try:
				async with client._lock:
					rotated = await rotate_1psidts(client.client, getattr(client, "verbose", False))
				if rotated:
					self._observe_rotation(client)
					logger.info("Managed Cookie rotation completed despite account-status probe=%s", getattr(getattr(client, "account_status", None), "name", "unknown"))
			except asyncio.CancelledError:
				raise
			except Exception as e:
				logger.warning("Managed Cookie rotation failed: %s", e)

	async def _supervisor(self) -> None:
		"""常驻监督循环：保活 + 自愈。"""
		logger.info("Gemini session supervisor loop running")
		while not self._closing:
			try:
				await asyncio.sleep(self.health_interval)
			except asyncio.CancelledError:
				raise

			if self._closing:
				break

			try:
				await self._supervise_once()
			except asyncio.CancelledError:
				raise
			except Exception as e:
				logger.warning("Supervisor tick failed: %s", e, exc_info=True)

	async def _supervise_once(self) -> None:
		client = self.client

		if client is None:
			# 还没有可用会话：按退避策略尝试初始化，避免密集重试
			now = time.time()
			if now >= self._next_init_at:
				try:
					await self.get_client()
				except Exception as e:
					self._last_error = str(e)
					self._next_init_at = now + self.init_retry_interval
					logger.warning("Supervisor could not (re)initialize Gemini client: %s", e)
			return

		reason = self.session_transport_problem(client)
		if reason:
			logger.warning("Gemini client transport became unusable (%s); rebuilding", reason)
			async with self.lock:
				await self._drop_client_locked(reason)
			try:
				await self.get_client()
			except Exception as e:
				self._last_error = str(e)
				self._next_init_at = time.time() + self.init_retry_interval
				logger.warning("Rebuild after session loss did not succeed: %s", e)
			return

		self._observe_rotation(client)
		await self._rearm_background_tasks(client)

	def _observe_rotation(self, client: GeminiClient) -> None:
		"""读取活跃会话里的 ``__Secure-1PSIDTS``，发现变化就落盘并触发备份。

		gemini-webapi 的 ``rotate_1psidts`` 会把轮换结果直接写回 ``client.cookies``，
		所以这里读到的就是当前真实生效的值。
		"""
		live_psidts = extract_cookie(getattr(client, "cookies", None), PSIDTS_COOKIE)
		if not live_psidts or live_psidts == self._last_1psidts:
			return

		previous = self._last_1psidts
		self._last_1psidts = live_psidts
		self._last_psidts_change_at = time.time()
		self._rotation_count += 1

		logger.info(
			"__Secure-1PSIDTS rotated in the background (%s -> %s)",
			mask_secret(previous) or "(none)",
			mask_secret(live_psidts),
		)

		psid, _ = self.credential_provider()
		if self.store.write_1psidts(psid, live_psidts):
			logger.info("Persisted the rotated __Secure-1PSIDTS to disk")

		if self.backup is not None and self.backup.enabled:
			asyncio.create_task(self._push_backup(psid, live_psidts))

	async def _rearm_background_tasks(self, client: GeminiClient) -> None:
		"""把 gemini-webapi 的后台轮换任务和活跃度看门狗重新拉起来。

		``start_auto_refresh`` 在账号状态不是 ``AVAILABLE`` 时会直接退出并把
		``refresh_task`` 置空，此后没有任何机制会再拉起它。``rearm_cooldown`` 用来
		避免状态持续异常时的忙循环——每轮最多重启一次。
		"""
		if not getattr(client, "auto_refresh", False) or not getattr(client, "_running", False):
			return

		# gemini-webapi 2.1.x only starts these tasks when its auxiliary account
		# status RPC says AVAILABLE. That RPC can be false-negative while the main
		# generation endpoint still works, so the supervisor owns the re-arm decision.

		now = time.time()
		if now < self._rearm_until:
			return

		if client.refresh_task is None or client.refresh_task.done():
			client.refresh_task = asyncio.create_task(client.start_auto_refresh())
			self._rearm_until = now + self.rearm_cooldown
			self._rearm_count += 1
			logger.info(
				"Re-armed the __Secure-1PSIDTS rotation task (account status: %s)",
				getattr(getattr(client, "account_status", None), "name", "unknown"),
			)
			return

		if client.activity_task is None or client.activity_task.done():
			client.activity_task = asyncio.create_task(client.start_activity_watchdog())
			self._rearm_until = now + self.rearm_cooldown
			self._rearm_count += 1
			logger.info("Re-armed the Gemini activity watchdog")

	async def _push_backup(self, psid: str, value: str) -> None:
		if self.backup is None:
			return
		try:
			result = await self.backup.push(psid, value)
			if result and not result.get("skipped"):
				logger.info("Cookie backup result: %s", result)
		except Exception as e:
			logger.warning("Cookie backup failed: %s", e)