diff --git "a/\347\272\242\345\214\205/README.md" "b/\347\272\242\345\214\205/README.md"
new file mode 100644
index 00000000..54a00bd6
--- /dev/null
+++ "b/\347\272\242\345\214\205/README.md"
@@ -0,0 +1,33 @@
+# 红包
+
+ToolDelta 类式插件。玩家可以使用经济计分板发送拼手气口令红包。
+
+## 使用
+
+```text
+.fhb <总金额> <份数> <口令>
+.发红包 <总金额> <份数> <口令>
+```
+
+缺少参数时，插件会在聊天栏逐项询问。输入 `取消` 或 `q` 可以退出。
+
+- 红包会向全服展示发送者、总金额和领取口令。
+- 领取成功消息始终全服播报。
+- 红包被抢完后会全服播报领取金额最高的玩家。
+- 每名玩家在同一个红包中只能领取一次，发送者也可以领取。
+- 红包 10 分钟未领完时，剩余金额自动退回；重启 ToolDelta 不会重置计时。
+- `�`、控制符、不可见格式符等无效字符不能用于口令；尝试使用时会收到禁止提示。
+
+## 配置
+
+首次加载后会生成 `插件配置文件/红包.json`：
+
+```json
+{
+  "经济计分板": "money",
+  "货币名称": "星尘",
+  "红包有效期秒数": 600
+}
+```
+
+需要安装 `聊天栏菜单` 0.4.1 或更高版本。
diff --git "a/\347\272\242\345\214\205/__init__.py" "b/\347\272\242\345\214\205/__init__.py"
new file mode 100644
index 00000000..9cb874ea
--- /dev/null
+++ "b/\347\272\242\345\214\205/__init__.py"
@@ -0,0 +1,182 @@
+"""ToolDelta 拼手气红包插件。"""
+from __future__ import annotations
+
+import threading
+from typing import Any
+
+from tooldelta import Chat, FrameExit, Player, Plugin, plugin_entry, utils
+
+from .runtime_reload import reload_plugin_modules
+
+reload_plugin_modules(__name__)
+
+from .broadcasting import RedPacketBroadcastMixin  # noqa: E402
+from .configuration import load_configuration  # noqa: E402
+from .interaction import CommandCollector  # noqa: E402
+from .messages import is_red_packet_command  # noqa: E402
+from .models import RedPacketState, player_identity  # noqa: E402
+from .service import RedPacketService  # noqa: E402
+from .storage import RedPacketStore  # noqa: E402
+from .text_validation import (  # noqa: E402
+    INVALID_CHARACTER_MESSAGE,
+    has_invalid_characters,
+)
+
+
+class LuckyRedPacket(RedPacketBroadcastMixin, Plugin):
+    """让玩家发送并领取拼手气口令红包。"""
+
+    name = "红包"
+    author = "Q3CC"
+    version = (0, 2, 3)
+    description = "支持全服口令领取的拼手气红包"
+
+    def __init__(self, frame: Any) -> None:
+        """初始化配置、持久化状态和事件监听器。"""
+        super().__init__(frame)
+        (
+            self.config,
+            self.scoreboard_name,
+            self.currency_name,
+            self.expiry_seconds,
+        ) = load_configuration(
+            self.name,
+            self.version,
+        )
+        self._prompting_players: set[str] = set()
+        self._prompting_lock = threading.Lock()
+        self._stop_event = threading.Event()
+        self._storage_ready = True
+        store = RedPacketStore(self.data_path / "红包状态.json")
+        try:
+            state = store.load()
+        except ValueError as err:
+            self._storage_ready = False
+            state = RedPacketState()
+            self.print_err(f"{err}；为防止资产异常，红包功能已停用")
+        self.service = RedPacketService(self, store, state)
+        self.ListenPreload(self.on_preload)
+        self.ListenActive(self.on_active)
+        self.ListenChat(self.on_chat)
+        self.ListenPlayerJoin(self.on_player_join)
+        self.ListenFrameExit(self.on_frame_exit)
+
+    def on_preload(self) -> None:
+        """注册聊天栏红包命令。"""
+        chatbar = self.GetPluginAPI("聊天栏菜单", (0, 4, 1))
+        chatbar.add_new_trigger(
+            ["fhb", "发红包"],
+            [
+                ("总金额", str, ""),
+                ("份数", str, ""),
+                ("口令", str, ""),
+            ],
+            "发送拼手气红包",
+            self.on_red_packet_command,
+        )
+
+    def on_active(self) -> None:
+        """恢复到期任务并启动退款检查线程。"""
+        if not self._storage_ready:
+            return
+        self.service.process_expired()
+        utils.createThread(self._expiry_loop, usage="红包到期退款")
+        self.print_suc(
+            f"红包插件已启用，货币为 {self.currency_name}，"
+            f"有效期 {self.expiry_seconds} 秒"
+        )
+
+    def on_frame_exit(self, _: FrameExit) -> None:
+        """通知后台退款线程停止。"""
+        self._stop_event.set()
+
+    def on_red_packet_command(
+        self,
+        player: Player,
+        args: tuple[Any, ...],
+    ) -> bool:
+        """接收红包命令并立即释放聊天栏菜单回调。"""
+        if not self._storage_ready:
+            player.show("§c红包数据异常，功能暂时不可用，请联系管理员§r")
+            return True
+        identity = player_identity(player)
+        with self._prompting_lock:
+            if identity in self._prompting_players:
+                player.show("§e你正在创建红包，请先完成当前输入或发送 q 取消§r")
+                return True
+            self._prompting_players.add(identity)
+        utils.createThread(
+            lambda: self._run_red_packet_command(player, args, identity),
+            usage=f"玩家 {player.name} 创建红包",
+        )
+        return True
+
+    def _run_red_packet_command(
+        self,
+        player: Player,
+        args: tuple[Any, ...],
+        identity: str,
+    ) -> None:
+        """在独立线程中完成缺参询问和红包创建。"""
+        collector = CommandCollector(
+            player,
+            args,
+            lambda _: None,
+        )
+        try:
+            request = collector.collect()
+            if request is not None:
+                if not request.has_valid_phrase():
+                    player.show(f"§c{INVALID_CHARACTER_MESSAGE}§r")
+                    return
+                self.service.create(player, request)
+        except Exception as err:
+            self.print_err(f"玩家 {player.name} 创建红包失败: {err}")
+            player.show("§c创建红包时发生错误，请稍后重试§r")
+        finally:
+            self._set_prompting(identity, False)
+
+    def on_chat(self, chat: Chat) -> None:
+        """把命中的普通聊天口令交给领取流程。"""
+        if not self._storage_ready:
+            return
+        raw_message = str(chat.msg)
+        phrase = raw_message.strip()
+        if is_red_packet_command(phrase):
+            return
+        if has_invalid_characters(raw_message):
+            chat.player.show(f"§c{INVALID_CHARACTER_MESSAGE}§r")
+            return
+        if not phrase:
+            return
+        identity = player_identity(chat.player)
+        with self._prompting_lock:
+            if identity in self._prompting_players:
+                return
+        if not self.service.has_active_phrase(phrase):
+            return
+        utils.createThread(
+            lambda: self.service.claim(chat.player, phrase),
+            usage=f"玩家 {chat.player.name} 领取红包",
+        )
+
+    def on_player_join(self, player: Player) -> None:
+        """向上线玩家发送待投递的退款通知。"""
+        if self._storage_ready:
+            self.service.deliver_refund_notices(player)
+
+    def _expiry_loop(self) -> None:
+        """周期检查红包到期状态。"""
+        while not self._stop_event.wait(1):
+            self.service.process_expired()
+
+    def _set_prompting(self, identity: str, prompting: bool) -> None:
+        """更新玩家是否正在回答创建问题。"""
+        with self._prompting_lock:
+            if prompting:
+                self._prompting_players.add(identity)
+            else:
+                self._prompting_players.discard(identity)
+
+
+entry = plugin_entry(LuckyRedPacket)
diff --git "a/\347\272\242\345\214\205/broadcasting.py" "b/\347\272\242\345\214\205/broadcasting.py"
new file mode 100644
index 00000000..3dfe4eb7
--- /dev/null
+++ "b/\347\272\242\345\214\205/broadcasting.py"
@@ -0,0 +1,48 @@
+"""红包全服公告发送。"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from .messages import (
+    claim_message,
+    completed_message,
+    created_message,
+    tellraw_payload,
+)
+from .models import RedPacket
+
+
+class RedPacketBroadcastMixin:
+    """为红包插件提供全服公告方法。"""
+
+    currency_name: str
+    game_ctrl: Any
+
+    def broadcast_created(self, packet: RedPacket) -> None:
+        """广播新创建的红包。"""
+        self._broadcast_raw(created_message(packet, self.currency_name))
+
+    def broadcast_claim(
+        self,
+        claimant_name: str,
+        sender_name: str,
+        amount: int,
+    ) -> None:
+        """广播玩家领取红包的结果。"""
+        self._broadcast_raw(
+            claim_message(
+                claimant_name,
+                sender_name,
+                amount,
+                self.currency_name,
+            )
+        )
+
+    def broadcast_completed(self, packet: RedPacket) -> None:
+        """广播红包抢完及手气最佳结果。"""
+        self._broadcast_raw(completed_message(packet, self.currency_name))
+
+    def _broadcast_raw(self, message: str) -> None:
+        """安全序列化并向全服发送消息。"""
+        self.game_ctrl.sendwocmd(f"/tellraw @a {tellraw_payload(message)}")
diff --git "a/\347\272\242\345\214\205/configuration.py" "b/\347\272\242\345\214\205/configuration.py"
new file mode 100644
index 00000000..b74b4aa6
--- /dev/null
+++ "b/\347\272\242\345\214\205/configuration.py"
@@ -0,0 +1,48 @@
+"""红包插件配置加载与校验。"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from tooldelta import cfg
+
+from .text_validation import has_invalid_characters
+
+
+DEFAULT_CONFIG = {
+    "经济计分板": "money",
+    "货币名称": "星尘",
+    "红包有效期秒数": 600,
+}
+CONFIG_STANDARD = {
+    "经济计分板": str,
+    "货币名称": str,
+    "红包有效期秒数": cfg.PInt,
+}
+
+
+def load_configuration(
+    plugin_name: str,
+    plugin_version: tuple[int, int, int],
+) -> tuple[dict[str, Any], str, str, int]:
+    """读取并规范化红包插件配置。"""
+    config, _ = cfg.get_plugin_config_and_version(
+        plugin_name,
+        CONFIG_STANDARD,
+        DEFAULT_CONFIG,
+        plugin_version,
+    )
+    scoreboard_name = str(config["经济计分板"]).strip()
+    currency_name = str(config["货币名称"]).strip()
+    expiry_seconds = int(config["红包有效期秒数"])
+    if (
+        not scoreboard_name
+        or any(character.isspace() for character in scoreboard_name)
+        or has_invalid_characters(scoreboard_name)
+    ):
+        raise ValueError("经济计分板不能为空、包含空格或无效字符")
+    if not currency_name or has_invalid_characters(currency_name):
+        raise ValueError("货币名称不能为空或包含无效字符")
+    if expiry_seconds < 1:
+        raise ValueError("红包有效期秒数必须大于 0")
+    return config, scoreboard_name, currency_name, expiry_seconds
diff --git "a/\347\272\242\345\214\205/datas.json" "b/\347\272\242\345\214\205/datas.json"
new file mode 100644
index 00000000..6a01565e
--- /dev/null
+++ "b/\347\272\242\345\214\205/datas.json"
@@ -0,0 +1,10 @@
+{
+  "plugin-id": "red_packet",
+  "author": "Q3CC",
+  "version": "0.2.3",
+  "description": "支持全服口令领取及手气最佳播报，未领完金额自动退回。",
+  "plugin-type": "classic",
+  "pre-plugins": {
+    "聊天栏菜单": "0.4.1"
+  }
+}
diff --git "a/\347\272\242\345\214\205/economy.py" "b/\347\272\242\345\214\205/economy.py"
new file mode 100644
index 00000000..a0a86186
--- /dev/null
+++ "b/\347\272\242\345\214\205/economy.py"
@@ -0,0 +1,63 @@
+"""经济计分板读写网关。"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from .models import player_identity
+
+
+class ScoreboardEconomy:
+    """集中封装带结果校验的计分板余额变更。"""
+
+    def __init__(self, plugin: Any) -> None:
+        """保存插件实例以访问计分板和在线玩家。"""
+        self.plugin = plugin
+
+    def get_balance(self, player: Any) -> int | None:
+        """读取玩家余额，失败时向玩家说明原因。"""
+        try:
+            return int(player.getScore(self.plugin.scoreboard_name, timeout=5))
+        except TimeoutError:
+            player.show("§c查询余额超时，请稍后重试§r")
+        except (ValueError, TypeError):
+            player.show(
+                f"§c无法读取经济计分板 §e「{self.plugin.scoreboard_name}」"
+                "§c，请联系管理员§r"
+            )
+        except Exception as err:
+            self.plugin.print_war(f"查询玩家 {player.name} 余额失败: {err}")
+            player.show("§c查询余额失败，请稍后重试§r")
+        return None
+
+    def change_score(self, target: str, amount: int) -> bool:
+        """增减目标分数并返回命令是否成功。"""
+        action = "add" if amount >= 0 else "remove"
+        value = abs(amount)
+        command = (
+            f"/scoreboard players {action} {target} "
+            f"{self.plugin.scoreboard_name} {value}"
+        )
+        try:
+            result = self.plugin.game_ctrl.sendwscmd_with_resp(command, 5)
+            return bool(getattr(result, "SuccessCount", 0))
+        except Exception as err:
+            self.plugin.print_war(f"执行经济计分板命令失败: {err}")
+            return False
+
+    def find_online_player(self, identity: str) -> Any | None:
+        """按稳定身份键查找在线玩家。"""
+        try:
+            players = self.plugin.game_ctrl.players.getAllPlayers()
+        except Exception:
+            return None
+        return next(
+            (player for player in players if player_identity(player) == identity),
+            None,
+        )
+
+    @staticmethod
+    def quoted_name(name: str) -> str:
+        """转义并引用离线计分板玩家名。"""
+        escaped = name.replace("\\", "\\\\").replace('"', '\\"')
+        return f'"{escaped}"'
diff --git "a/\347\272\242\345\214\205/interaction.py" "b/\347\272\242\345\214\205/interaction.py"
new file mode 100644
index 00000000..dd139cea
--- /dev/null
+++ "b/\347\272\242\345\214\205/interaction.py"
@@ -0,0 +1,126 @@
+"""红包命令参数校验和缺参询问。"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import Any
+
+from .models import MAX_SCORE, RedPacketRequest
+from .text_validation import require_valid_characters
+
+
+CANCEL_WORDS = {"q", "取消"}
+
+
+class CommandCollector:
+    """把聊天栏菜单传入的字符串参数整理成创建请求。"""
+
+    def __init__(
+        self,
+        player: Any,
+        args: tuple[Any, ...],
+        set_prompting: Callable[[bool], None],
+    ) -> None:
+        """保存玩家和最多三个命令参数。"""
+        values = [str(value) for value in args[:3]]
+        self.values = values + [""] * (3 - len(values))
+        self.player = player
+        self.set_prompting = set_prompting
+
+    def collect(self) -> RedPacketRequest | None:
+        """校验已有参数并逐项询问缺失值。"""
+        interactive = any(value == "" for value in self.values)
+        if interactive:
+            self.set_prompting(True)
+        try:
+            amount = self._value(
+                0,
+                self._parse_amount,
+                "§e请输入红包总金额（正整数）：§r",
+            )
+            if amount is None:
+                return None
+            count = self._value(
+                1,
+                lambda value: self._parse_count(value, amount),
+                f"§e请输入红包份数（1～{amount}）：§r",
+            )
+            if count is None:
+                return None
+            phrase = self._value(
+                2,
+                self._parse_phrase,
+                "§e请输入红包口令（单个词）：§r",
+            )
+            if phrase is None:
+                return None
+            return RedPacketRequest(amount, count, phrase)
+        finally:
+            if interactive:
+                self.set_prompting(False)
+
+    def _value(
+        self,
+        index: int,
+        parser: Callable[[str], Any],
+        prompt: str,
+    ) -> Any | None:
+        """取得并解析一个参数，交互输入失败时允许重试。"""
+        supplied = self.values[index]
+        if supplied != "":
+            try:
+                return parser(require_valid_characters(supplied))
+            except ValueError as err:
+                self.player.show(f"§c{err}§r")
+                return None
+        while True:
+            response = self.player.input(prompt, timeout=60)
+            if response is None:
+                self.player.show("§c等待输入超时，已取消发送红包§r")
+                return None
+            response = str(response)
+            try:
+                require_valid_characters(response)
+            except ValueError as err:
+                self.player.show(f"§c{err}§r")
+                continue
+            response = response.strip()
+            if response.casefold() in CANCEL_WORDS:
+                self.player.show("§7已取消发送红包§r")
+                return None
+            try:
+                return parser(response)
+            except ValueError as err:
+                self.player.show(f"§c{err}§r")
+
+    @staticmethod
+    def _parse_amount(value: str) -> int:
+        """解析红包总金额。"""
+        try:
+            amount = int(value)
+        except ValueError as err:
+            raise ValueError("红包金额必须是整数") from err
+        if not 1 <= amount <= MAX_SCORE:
+            raise ValueError(f"红包金额必须在 1～{MAX_SCORE} 之间")
+        return amount
+
+    @staticmethod
+    def _parse_count(value: str, amount: int) -> int:
+        """解析红包份数并确保总金额足够分配。"""
+        try:
+            count = int(value)
+        except ValueError as err:
+            raise ValueError("红包份数必须是整数") from err
+        if not 1 <= count <= amount:
+            raise ValueError(f"红包份数必须在 1～{amount} 之间")
+        return count
+
+    @staticmethod
+    def _parse_phrase(value: str) -> str:
+        """解析单词形式的红包口令。"""
+        phrase = value.strip()
+        if not 1 <= len(phrase) <= 32:
+            raise ValueError("红包口令长度必须为 1～32 个字符")
+        if any(character.isspace() for character in phrase):
+            raise ValueError("红包口令不能包含空格")
+        return phrase
diff --git "a/\347\272\242\345\214\205/messages.py" "b/\347\272\242\345\214\205/messages.py"
new file mode 100644
index 00000000..de00eedd
--- /dev/null
+++ "b/\347\272\242\345\214\205/messages.py"
@@ -0,0 +1,75 @@
+"""红包聊天文案和 tellraw 序列化。"""
+
+from __future__ import annotations
+
+import json
+
+from .models import RedPacket
+
+
+def created_message(packet: RedPacket, currency_name: str) -> str:
+    """生成红包创建公告。"""
+    return (
+        "§c§l------->>抢红包<<-------§r\n"
+        f"§f玩家 §b{packet.sender_name} §f发送了价值 "
+        f"§6§l{packet.total_amount} {currency_name}§r§f 的红包\n"
+        f"§f 发送口令 §a§l「{packet.phrase}」§r§f 领取\n"
+        "§c----------------------------§r"
+    )
+
+
+def claim_message(
+    claimant_name: str,
+    sender_name: str,
+    amount: int,
+    currency_name: str,
+) -> str:
+    """生成单次领取公告。"""
+    return (
+        f"§f玩家 §a§l{claimant_name}§r§f 从 §b{sender_name} "
+        f"§f的红包中抢到了 §6§l{amount} {currency_name}§r"
+    )
+
+
+def completed_message(packet: RedPacket, currency_name: str) -> str:
+    """生成红包抢完及手气最佳公告。"""
+    return (
+        f"§f玩家 §b{packet.sender_name} §f发送的红包被抢完，"
+        f"§a§l{packet.best_claimant_name}§r§f 是手气最佳，抢到了 "
+        f"§6§l{packet.best_claim_amount} {currency_name}§r"
+    )
+
+
+def refund_message(packet: RedPacket, currency_name: str) -> str:
+    """生成红包到期退款通知。"""
+    packet_label = (
+        f"红包「{packet.phrase}」"
+        if packet.has_valid_phrase()
+        else "含无效字符的红包"
+    )
+    return (
+        f"§e{packet_label} 已过期，剩余 "
+        f"§6§l{packet.remaining_amount} {currency_name}§r§e 已退回§r"
+    )
+
+
+def tellraw_payload(message: str) -> str:
+    """把消息安全序列化为 tellraw JSON。"""
+    return json.dumps(
+        {"rawtext": [{"text": message}]},
+        ensure_ascii=False,
+        separators=(",", ":"),
+    )
+
+
+def is_red_packet_command(message: str) -> bool:
+    """判断聊天内容是否为红包命令。"""
+    lowered = message.casefold()
+    return lowered == ".fhb" or lowered.startswith(".fhb ") or (
+        message == ".发红包"
+        or message.startswith(".发红包 ")
+        or message == "。发红包"
+        or message.startswith("。发红包 ")
+        or lowered == "。fhb"
+        or lowered.startswith("。fhb ")
+    )
diff --git "a/\347\272\242\345\214\205/models.py" "b/\347\272\242\345\214\205/models.py"
new file mode 100644
index 00000000..be1a00ea
--- /dev/null
+++ "b/\347\272\242\345\214\205/models.py"
@@ -0,0 +1,168 @@
+"""红包领域模型和纯计算函数。"""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass, field
+from random import Random
+from typing import Any
+
+from .text_validation import has_invalid_characters
+
+
+MAX_SCORE = 2_147_483_647
+
+
+@dataclass(frozen=True)
+class RedPacketRequest:
+    """玩家创建红包时已经校验过的参数。"""
+
+    total_amount: int
+    total_count: int
+    phrase: str
+
+    def has_valid_phrase(self) -> bool:
+        """检查创建请求中的口令是否安全。"""
+        return (
+            1 <= len(self.phrase) <= 32
+            and not any(character.isspace() for character in self.phrase)
+            and not has_invalid_characters(self.phrase)
+        )
+
+
+@dataclass
+class RedPacket:
+    """一个尚未完成退款或领取的红包。"""
+
+    packet_id: str
+    sender_name: str
+    sender_key: str
+    total_amount: int
+    remaining_amount: int
+    total_count: int
+    remaining_count: int
+    phrase: str
+    is_public: bool
+    claimed_keys: list[str] = field(default_factory=list)
+    best_claimant_name: str = ""
+    best_claim_amount: int = 0
+    created_at: float = 0.0
+    expires_at: float = 0.0
+    status: str = "active"
+
+    def to_dict(self) -> dict[str, Any]:
+        """转换为可持久化字典。"""
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> "RedPacket":
+        """从持久化字典恢复并校验红包。"""
+        packet = cls(
+            packet_id=str(data["packet_id"]),
+            sender_name=str(data["sender_name"]),
+            sender_key=str(data["sender_key"]),
+            total_amount=int(data["total_amount"]),
+            remaining_amount=int(data["remaining_amount"]),
+            total_count=int(data["total_count"]),
+            remaining_count=int(data["remaining_count"]),
+            phrase=str(data["phrase"]),
+            is_public=bool(data["is_public"]),
+            claimed_keys=[str(item) for item in data.get("claimed_keys", [])],
+            best_claimant_name=str(data.get("best_claimant_name", "")),
+            best_claim_amount=int(data.get("best_claim_amount", 0)),
+            created_at=float(data["created_at"]),
+            expires_at=float(data["expires_at"]),
+            status=str(data.get("status", "active")),
+        )
+        unsafe_phrase = not packet.has_valid_phrase()
+        packet.validate(allow_unsafe_phrase=unsafe_phrase)
+        if unsafe_phrase:
+            packet.status = "refund_pending"
+        return packet
+
+    def has_valid_phrase(self) -> bool:
+        """检查持久化红包口令是否安全。"""
+        return (
+            1 <= len(self.phrase) <= 32
+            and not any(character.isspace() for character in self.phrase)
+            and not has_invalid_characters(self.phrase)
+        )
+
+    def validate(self, allow_unsafe_phrase: bool = False) -> None:
+        """校验红包金额、份数、口令和状态。"""
+        if not self.packet_id or not self.sender_name or not self.sender_key:
+            raise ValueError("红包身份信息不完整")
+        if self.total_amount < 1 or self.total_amount > MAX_SCORE:
+            raise ValueError("红包总金额无效")
+        if self.total_count < 1 or self.total_count > self.total_amount:
+            raise ValueError("红包总份数无效")
+        if not 0 <= self.remaining_amount <= self.total_amount:
+            raise ValueError("红包剩余金额无效")
+        if not 0 <= self.remaining_count <= self.total_count:
+            raise ValueError("红包剩余份数无效")
+        if not 0 <= self.best_claim_amount <= self.total_amount:
+            raise ValueError("红包手气最佳金额无效")
+        if bool(self.best_claimant_name) != bool(self.best_claim_amount):
+            raise ValueError("红包手气最佳信息不完整")
+        if not allow_unsafe_phrase and not self.has_valid_phrase():
+            raise ValueError("红包口令无效")
+        if self.status not in {"active", "refund_pending"}:
+            raise ValueError("红包状态无效")
+
+
+@dataclass
+class RedPacketState:
+    """需要持久化的全部红包状态。"""
+
+    packets: dict[str, RedPacket] = field(default_factory=dict)
+    refund_notices: dict[str, list[str]] = field(default_factory=dict)
+
+    def to_dict(self) -> dict[str, Any]:
+        """转换全部红包状态为持久化字典。"""
+        return {
+            "format_version": 1,
+            "packets": [
+                packet.to_dict() for packet in self.packets.values()
+            ],
+            "refund_notices": self.refund_notices,
+        }
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> "RedPacketState":
+        """从持久化字典恢复全部红包状态。"""
+        if int(data.get("format_version", 0)) != 1:
+            raise ValueError("不支持的红包数据版本")
+        packets = {}
+        for raw_packet in data.get("packets", []):
+            packet = RedPacket.from_dict(raw_packet)
+            if packet.packet_id in packets:
+                raise ValueError("存在重复的红包 ID")
+            packets[packet.packet_id] = packet
+        notices = {
+            str(key): [str(message) for message in messages]
+            for key, messages in data.get("refund_notices", {}).items()
+        }
+        return cls(packets=packets, refund_notices=notices)
+
+
+def choose_lucky_amount(
+    remaining_amount: int,
+    remaining_count: int,
+    rng: Random,
+) -> int:
+    """使用双均值法抽取一份，并保证每个剩余红包至少有 1。"""
+    if remaining_count < 1 or remaining_amount < remaining_count:
+        raise ValueError("红包剩余金额和份数不合法")
+    if remaining_count == 1:
+        return remaining_amount
+    guaranteed_max = remaining_amount - (remaining_count - 1)
+    double_mean = max(1, (remaining_amount * 2) // remaining_count)
+    return rng.randint(1, min(guaranteed_max, double_mean))
+
+
+def player_identity(player: Any) -> str:
+    """获取跨重启尽量稳定的玩家身份键。"""
+    for prefix, attribute in (("xuid", "xuid"), ("uuid", "uuid")):
+        value = str(getattr(player, attribute, "") or "").strip()
+        if value:
+            return f"{prefix}:{value}"
+    return f"name:{str(getattr(player, 'name', player))}"
diff --git "a/\347\272\242\345\214\205/runtime_reload.py" "b/\347\272\242\345\214\205/runtime_reload.py"
new file mode 100644
index 00000000..7c4c6471
--- /dev/null
+++ "b/\347\272\242\345\214\205/runtime_reload.py"
@@ -0,0 +1,28 @@
+"""ToolDelta 热重载时同步刷新红包子模块。"""
+
+from __future__ import annotations
+
+import importlib
+import sys
+
+
+RELOAD_ORDER = (
+    "text_validation",
+    "models",
+    "configuration",
+    "interaction",
+    "economy",
+    "storage",
+    "messages",
+    "broadcasting",
+    "service",
+)
+
+
+def reload_plugin_modules(package_name: str) -> None:
+    """按依赖顺序刷新已缓存的红包模块。"""
+    for module_suffix in RELOAD_ORDER:
+        module_name = f"{package_name}.{module_suffix}"
+        cached_module = sys.modules.get(module_name)
+        if cached_module is not None:
+            importlib.reload(cached_module)
diff --git "a/\347\272\242\345\214\205/service.py" "b/\347\272\242\345\214\205/service.py"
new file mode 100644
index 00000000..748a6e6d
--- /dev/null
+++ "b/\347\272\242\345\214\205/service.py"
@@ -0,0 +1,200 @@
+"""红包创建、领取和退款业务。"""
+
+from __future__ import annotations
+
+import random
+import threading
+import time
+import uuid
+from typing import Any
+
+from . import models
+from .economy import ScoreboardEconomy
+from .messages import refund_message
+from .storage import RedPacketStore
+
+
+class RedPacketService:
+    """串行处理所有会改变红包或计分板余额的操作。"""
+
+    def __init__(
+        self,
+        plugin: Any,
+        store: RedPacketStore,
+        state: models.RedPacketState,
+        rng: random.Random | None = None,
+    ) -> None:
+        """保存业务依赖并创建跨操作线程锁。"""
+        self.plugin = plugin
+        self.store = store
+        self.state = state
+        self.rng = rng or random.SystemRandom()
+        self.lock = threading.RLock()
+        self.economy = ScoreboardEconomy(plugin)
+
+    def has_active_phrase(self, phrase: str) -> bool:
+        """检查口令是否对应活跃红包。"""
+        with self.lock:
+            return self._find_active(phrase) is not None
+
+    def create(self, player: Any, request: models.RedPacketRequest) -> bool:
+        """扣款、保存并激活玩家创建的红包。"""
+        with self.lock:
+            if not request.has_valid_phrase():
+                player.show("§c禁止使用无效字符§r")
+                return False
+            if self._find_active(request.phrase) is not None:
+                player.show("§e这个口令已有未结束的红包，请换一个口令§r")
+                return False
+            balance = self.economy.get_balance(player)
+            if balance is None:
+                return False
+            if balance < request.total_amount:
+                currency = self.plugin.currency_name
+                player.show(
+                    f"§c余额不足：需要 §6{request.total_amount} {currency}"
+                    f"§c，你只有 §6{balance} {currency}§r"
+                )
+                return False
+            if not self.economy.change_score(
+                player.safe_name,
+                -request.total_amount,
+            ):
+                player.show("§c扣除余额失败，红包没有创建，请稍后重试§r")
+                return False
+            now = time.time()
+            packet = models.RedPacket(
+                packet_id=uuid.uuid4().hex,
+                sender_name=player.name,
+                sender_key=models.player_identity(player),
+                total_amount=request.total_amount,
+                remaining_amount=request.total_amount,
+                total_count=request.total_count,
+                remaining_count=request.total_count,
+                phrase=request.phrase,
+                is_public=True,
+                created_at=now,
+                expires_at=now + self.plugin.expiry_seconds,
+            )
+            self.state.packets[packet.packet_id] = packet
+            if not self._save():
+                self.state.packets.pop(packet.packet_id, None)
+                refunded = self.economy.change_score(
+                    player.safe_name,
+                    request.total_amount,
+                )
+                message = (
+                    "§c红包数据保存失败，金额已退回§r"
+                    if refunded
+                    else "§c红包数据保存和自动退款均失败，请立即联系管理员§r"
+                )
+                player.show(message)
+                return False
+            self.plugin.broadcast_created(packet)
+            return True
+
+    def claim(self, player: Any, phrase: str) -> bool:
+        """为玩家随机发放一份命中口令的红包。"""
+        with self.lock:
+            packet = self._find_active(phrase)
+            if packet is None:
+                return False
+            claimant_key = models.player_identity(player)
+            if claimant_key in packet.claimed_keys:
+                player.show("§e你已经领取过这个红包了§r")
+                return True
+            amount = models.choose_lucky_amount(
+                packet.remaining_amount,
+                packet.remaining_count,
+                self.rng,
+            )
+            if not self.economy.change_score(player.safe_name, amount):
+                player.show("§c红包领取失败，金额没有被消耗，请稍后重试§r")
+                return True
+            packet.remaining_amount -= amount
+            packet.remaining_count -= 1
+            packet.claimed_keys.append(claimant_key)
+            if amount > packet.best_claim_amount:
+                packet.best_claimant_name = player.name
+                packet.best_claim_amount = amount
+            completed = packet.remaining_count == 0
+            if completed:
+                self.state.packets.pop(packet.packet_id, None)
+            if not self._save():
+                self.plugin.print_err(
+                    f"玩家 {player.name} 已领取红包，但红包状态保存失败"
+                )
+            self.plugin.broadcast_claim(player.name, packet.sender_name, amount)
+            if completed:
+                self.plugin.broadcast_completed(packet)
+            return True
+
+    def process_expired(self, now: float | None = None) -> None:
+        """标记到期红包并重试所有待处理退款。"""
+        current_time = time.time() if now is None else now
+        with self.lock:
+            changed = False
+            for packet in self.state.packets.values():
+                if packet.status == "active" and packet.expires_at <= current_time:
+                    packet.status = "refund_pending"
+                    changed = True
+            if changed:
+                self._save()
+            for packet in list(self.state.packets.values()):
+                if packet.status == "refund_pending":
+                    self._try_refund(packet)
+
+    def deliver_refund_notices(self, player: Any) -> None:
+        """向上线玩家投递离线期间产生的退款通知。"""
+        key = models.player_identity(player)
+        with self.lock:
+            messages = self.state.refund_notices.pop(key, [])
+            if not messages:
+                return
+            if not self._save():
+                self.state.refund_notices[key] = messages
+                return
+        for message in messages:
+            player.show(message)
+
+    def _try_refund(self, packet: models.RedPacket) -> None:
+        """尝试退回一个红包的全部剩余金额。"""
+        online_player = self.economy.find_online_player(packet.sender_key)
+        target = (
+            online_player.safe_name
+            if online_player is not None
+            else self.economy.quoted_name(packet.sender_name)
+        )
+        if not self.economy.change_score(target, packet.remaining_amount):
+            self.plugin.print_war(
+                f"红包 {packet.packet_id} 退款失败，稍后将自动重试"
+            )
+            return
+        message = refund_message(packet, self.plugin.currency_name)
+        self.state.packets.pop(packet.packet_id, None)
+        if online_player is not None:
+            online_player.show(message)
+        else:
+            self.state.refund_notices.setdefault(packet.sender_key, []).append(
+                message
+            )
+        if not self._save():
+            self.plugin.print_err(
+                f"红包 {packet.packet_id} 已退款，但状态保存失败"
+            )
+
+    def _find_active(self, phrase: str) -> models.RedPacket | None:
+        """按大小写精确查找活跃红包。"""
+        for packet in self.state.packets.values():
+            if packet.status == "active" and packet.phrase == phrase:
+                return packet
+        return None
+
+    def _save(self) -> bool:
+        """保存当前状态并统一记录失败。"""
+        try:
+            self.store.save(self.state)
+            return True
+        except Exception as err:
+            self.plugin.print_err(f"保存红包数据失败: {err}")
+            return False
diff --git "a/\347\272\242\345\214\205/storage.py" "b/\347\272\242\345\214\205/storage.py"
new file mode 100644
index 00000000..26333f52
--- /dev/null
+++ "b/\347\272\242\345\214\205/storage.py"
@@ -0,0 +1,35 @@
+"""红包状态的安全 JSON 持久化。"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from tooldelta.utils.safe_json import safe_json_dump, safe_json_load
+
+from .models import RedPacketState
+
+
+class RedPacketStore:
+    """把红包状态保存到插件数据目录。"""
+
+    def __init__(self, path: Path) -> None:
+        """保存红包状态文件路径。"""
+        self.path = path
+
+    def load(self) -> RedPacketState:
+        """读取状态文件，不存在时创建空状态。"""
+        if not self.path.exists():
+            state = RedPacketState()
+            self.save(state)
+            return state
+        try:
+            raw_state = safe_json_load(str(self.path))
+            if not isinstance(raw_state, dict):
+                raise ValueError("红包数据根节点必须是对象")
+            return RedPacketState.from_dict(raw_state)
+        except (KeyError, TypeError, ValueError) as err:
+            raise ValueError(f"红包数据文件损坏: {err}") from err
+
+    def save(self, state: RedPacketState) -> None:
+        """以安全 JSON 写入完整红包状态。"""
+        safe_json_dump(state.to_dict(), self.path, indent=2)
diff --git "a/\347\272\242\345\214\205/text_validation.py" "b/\347\272\242\345\214\205/text_validation.py"
new file mode 100644
index 00000000..a36aa9be
--- /dev/null
+++ "b/\347\272\242\345\214\205/text_validation.py"
@@ -0,0 +1,25 @@
+"""用户可控文本的 Unicode 安全校验。"""
+
+from __future__ import annotations
+
+import unicodedata
+
+
+INVALID_CHARACTER_MESSAGE = "禁止使用无效字符"
+FORBIDDEN_UNICODE_CATEGORIES = frozenset({"Cc", "Cf", "Cs", "Co", "Cn"})
+
+
+def has_invalid_characters(text: str) -> bool:
+    """检查替换符、控制符、格式符、代理区、私用区和非字符。"""
+    return any(
+        character == "\ufffd"
+        or unicodedata.category(character) in FORBIDDEN_UNICODE_CATEGORIES
+        for character in text
+    )
+
+
+def require_valid_characters(text: str) -> str:
+    """返回安全文本；发现无效字符时仅给出固定错误，不回显输入。"""
+    if has_invalid_characters(text):
+        raise ValueError(INVALID_CHARACTER_MESSAGE)
+    return text