Release beta 1.1.2.1019

CHANGELOG.md

@@ -1,5 +1,18 @@
 # Changelog
 
+## beta 1.1.2.1019 - 2026-10-19 16:38:10
+
+- 新增低频主动插话、随机冷门知识问答和更新生命周期提示：Taffy 会偶尔自然冒泡，抛出限时答题奖励摩拉，并在停机、上线和更新就绪时主动说明状态。
+- Use recent safe group context to replace some hourly scan slots with an
+  ambient chat message during active daytime windows.
+- Add a 200+ question CS, Apex, Overwatch, and LoL trivia bank with
+  five-minute `/回答 A/B/C/D` answers, recent-question avoidance, and
+  rarity-based Mora rewards.
+- Limit ambient chat with deterministic probability, multi-hour cooldown, and
+  minimum recent-speaker activity so proactive replies feel occasional.
+- Send guarded shutdown and startup lifecycle notices during production
+  cutover before the normal ready-stage update announcement and compensation.
+
 ## beta 1.1.1.1016 - 2026-10-16 17:19:45
 
 - 修复扫码加权重复购买只刷新不叠加的问题。

README.md

@@ -8,7 +8,7 @@ JWeBot 是运行于 Linux 的单群微信自动回复系统。它通过本地 OC
 观察选定的微信窗口，将消息写入持久化 Edge 账本，调用经过审核的命令插件，
 并通过带视觉确认的 GUI 发送链路回复。
 
-当前版本：`beta 1.1.1.1016`，更新时间：2026-10-16 17:19:45。
+当前版本：`beta 1.1.2.1019`，更新时间：2026-10-19 16:38:10。
 
 1.0 Beta 收口了当前稳定功能：本地命令插件、人物与上下文索引、
 Steam 状态与时长查询、虚拟摩拉游戏、多人彩票、21 点、CS Major
@@ -89,7 +89,7 @@ PATH="$PWD/.venv/bin:$PATH" ./tools/check-local.sh
 /彩票 开始|报名|当前状态|开奖|取消（/lottery）
 /21点 开始 [1～500]；开局后 /21点 要牌|停牌|状态|结束
 /21点 多人 启动；/21点 加入；/21点 多人 开始；/21点 放200
-/摩拉（兼 /原石）；/原石祈愿；/原石榜；/扫码加权 人名；/Taffy发话 内容
+/摩拉（兼 /原石）；/原石祈愿；/原石榜；/扫码加权 人名（24小时+5）；/Taffy发话 内容
 /游戏记录 [页码]（/gamehistory）
 /游戏记录 [页码]（/gamehistory）
 /游戏帮助（/games）
@@ -269,10 +269,10 @@ visually confirmed GUI bridge.
 Current version:
 
 ```text
-beta 1.1.1.1016
+beta 1.1.2.1019
 ```
 
-Updated: 2026-10-16 17:19:45.
+Updated: 2026-10-19 16:38:10.
 
 The 1.0 Beta baseline consolidates the stable local command set, person and
 context indexes, Steam presence/playtime queries, virtual coin games,

VERSION

@@ -1 +1 @@
-beta 1.1.1.1016
+beta 1.1.2.1019

deploy/scripts/deploy-jwebot-production.sh

@@ -341,6 +341,19 @@ verify_edge_health() {
         'import json,sys; value=json.load(sys.stdin); raise SystemExit(0 if value.get("status") == "ok" else 1)'
 }
 
+send_release_lifecycle_notice() {
+  local stage=$1
+  /usr/sbin/runuser -u "${target_user}" -- /usr/bin/env \
+    HOME="${target_home}" \
+    XDG_RUNTIME_DIR="${user_runtime}" \
+    DBUS_SESSION_BUS_ADDRESS="unix:path=${user_runtime}/bus" \
+    "${runtime_python}" "${source_dir}/jwebot_release_update.py" \
+      --release-file "${source_dir}/VERSION" \
+      --database "${target_edge_state}/edge.sqlite3" \
+      --delivery-helper "${target_runtime}/jwebot_edge_delivery.py" \
+      --lifecycle-stage "${stage}"
+}
+
 render_root_configuration() {
   local output_dir="$1"
   sed \
@@ -471,6 +484,9 @@ jwebot_log "backup ready path=${backup_dir}"
 armed=1
 jwebot_progress 3 8 "stop services"
 jwebot_log "cutover starting"
+jwebot_log "release lifecycle shutdown notice starting"
+send_release_lifecycle_notice shutdown
+jwebot_log "release lifecycle shutdown notice ready"
 user_systemctl stop \
   wechat-ocr-adaptive-recorder.service \
   jwebot-edge.service \
@@ -609,6 +625,10 @@ if [[ -f "${target_edge_state}/edge.sqlite3" ]]; then
 fi
 jwebot_log "database ready quick_check=${db_check:-absent}"
 
+jwebot_log "release lifecycle startup notice starting"
+send_release_lifecycle_notice startup
+jwebot_log "release lifecycle startup notice ready"
+
 jwebot_progress 7 8 "verify backup"
 (
   cd "${backup_dir}"

docs/command-catalog.md

@@ -1,6 +1,6 @@
 # JWeBot Command Catalog
 
-Current baseline: `beta 1.1.1.1016`.
+Current baseline: `beta 1.1.2.1019`.
 
 This catalog records the reviewed commands shipped with the built-in local
 extensions. Runtime ownership is frozen at startup from the generated
@@ -48,9 +48,10 @@ descriptor inventory; plugin code delegates to reviewed service profiles.
 | `/还款 [金额|全部]` | `/repay [amount|all]` | Repay an active virtual coin loan. |
 | `/原石祈愿` | `/祈愿`, `/wish` | Spend 1 primogem to draw 20-10000 Mora with expected value 3600 Mora, with a chance to gain a scan-boost ticket. |
 | `/原石榜 [人数]` | `/primogem榜` | Show the group primogem leaderboard with source details. |
-| `/扫码加权 人名` | `/扫码概率 人名` | Spend a scan-boost ticket or 1 primogem to raise that person's hourly scan weight. |
+| `/扫码加权 人名` | `/扫码概率 人名` | Spend a scan-boost ticket or 1 primogem to give that person +5 hourly scan weight for 24 hours. |
 | `/Taffy发话 内容` | `/taffysay content` | Spend Taffy speech credit; 1 primogem converts to 2 credits when needed. |
 | `/Taffy发话 定时 HH:MM 内容` | - | Schedule a Taffy speech through the guarded clock delivery queue. |
+| `/回答 A` | `/answer A` | Answer the active five-minute cold-knowledge quiz for a rarity-based Mora reward. |
 | `/抽奖 [金额]` | `/draw` | Play a single-player prize draw. |
 | `/彩票 开始` | `/lottery start` | Open a multiplayer lottery round. |
 | `/彩票 报名 数字 金额` | `/lottery join` | Join or update a lottery entry. |

extensions/jwebot/trivia-quiz/README.md

@@ -0,0 +1,18 @@
+# trivia-quiz
+
+Author: `jwebot`
+
+冷门知识问答
+
+Help:
+
+- `/回答 A；/answer A`
+
+Registered commands:
+
+- `/回答`
+- `/answer`
+
+The adapter has no direct Ledger, GUI, environment, shell, or network access.
+It delegates to the reviewed `trivia-quiz` Kernel service profile
+and returns a structured `ReplyPlan`.

extensions/jwebot/trivia-quiz/plugin.toml

@@ -0,0 +1,14 @@
+id = "trivia-quiz"
+author = "jwebot"
+description = "冷门知识问答"
+help = ["/回答 A；/answer A"]
+version = "1.0.0"
+api_version = "1"
+entrypoint = "trivia_quiz:plugin"
+type = "command"
+
+[capabilities]
+commands = ["/回答", "/answer"]
+storage = false
+network_profiles = []
+service_profiles = ["trivia-quiz"]

extensions/jwebot/trivia-quiz/src/trivia_quiz.py

@@ -0,0 +1,45 @@
+"""Generated reviewed adapter for trivia-quiz."""
+
+from jwebot_sdk import Event, ExtensionResult, ReplyPlan, SingleGroupContext
+
+
+PROFILE = 'trivia-quiz'
+AUTHOR = 'jwebot'
+DESCRIPTION = '冷门知识问答'
+HELP_ENTRIES = ('/回答 A；/answer A',)
+COMMANDS = ('/回答', '/answer')
+
+
+def _handler(context: SingleGroupContext, command: str):
+    def handle(event: Event, arguments: str) -> ExtensionResult:
+        response = context.services.call(
+            PROFILE,
+            operation="execute",
+            payload={
+                "command": command,
+                "arguments": arguments,
+                "event": event.to_mapping(),
+            },
+        )
+        texts = response.get("texts")
+        if isinstance(texts, list) and texts and all(
+            isinstance(item, str) and item.strip() for item in texts
+        ):
+            return ExtensionResult(
+                reply=ReplyPlan.multi_text(texts),
+                metadata={"local_extension": PROFILE, "author": AUTHOR},
+            )
+        text = response.get("text")
+        if not isinstance(text, str) or not text.strip():
+            raise ValueError("local service returned invalid text")
+        return ExtensionResult(
+            reply=ReplyPlan.text(text),
+            metadata={"local_extension": PROFILE, "author": AUTHOR},
+        )
+
+    return handle
+
+
+def plugin(context: SingleGroupContext) -> None:
+    for command in COMMANDS:
+        context.commands.register(command, _handler(context, command))

extensions/jwebot/trivia-quiz/tests/test_plugin.py

@@ -0,0 +1,29 @@
+import sys
+import unittest
+from pathlib import Path
+from unittest import mock
+
+
+SOURCE = Path(__file__).resolve().parents[1] / "src"
+sys.path.insert(0, str(SOURCE))
+
+from trivia_quiz import AUTHOR, DESCRIPTION, HELP_ENTRIES, plugin
+
+
+class PluginTests(unittest.TestCase):
+    def test_registers_declared_commands_and_author(self):
+        context = mock.Mock()
+        plugin(context)
+        registered = [
+            call.args[0]
+            for call in context.commands.register.call_args_list
+        ]
+        self.assertEqual(AUTHOR, 'jwebot')
+        self.assertEqual(DESCRIPTION, '冷门知识问答')
+        self.assertEqual(HELP_ENTRIES, ('/回答 A；/answer A',))
+        self.assertIn('/回答', registered)
+        self.assertEqual(len(registered), 2)
+
+
+if __name__ == "__main__":
+    unittest.main()

jwebot_release_update.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Apply the mandatory post-release grant and group announcement."""
+"""Apply the mandatory release lifecycle notices, grant, and announcement."""
 
 from __future__ import annotations
 
@@ -39,6 +39,8 @@ PRIMOGEM_COMPENSATION_RELEASES = {
 MAX_BALANCE = 1_000_000_000
 MAX_SUMMARY_CHARS = 300
 MAX_ANNOUNCEMENT_CHARS = 900
+MAX_LIFECYCLE_CHARS = 300
+LIFECYCLE_STAGES = frozenset({"shutdown", "startup"})
 RELEASE_PATTERN = re.compile(
     r"^beta (?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)\.(?P<date>\d{4})$"
 )
@@ -140,6 +142,24 @@ def build_announcement(release: str, summary: str) -> str:
     return announcement
 
 
+def build_lifecycle_message(release: str, stage: str) -> str:
+    if stage == "shutdown":
+        message = (
+            f"Taffy 准备停机更新到 {release}，会短暂离线一下；"
+            "别慌，我很快回来喵"
+        )
+    elif stage == "startup":
+        message = (
+            f"Taffy 刚上线，正在做 {release} 的最后自检；"
+            "确认就绪后会发送更新内容和补偿说明喵"
+        )
+    else:
+        raise ReleaseUpdateError(f"unsupported lifecycle stage: {stage}")
+    if len(message) > MAX_LIFECYCLE_CHARS:
+        raise ReleaseUpdateError("release lifecycle message is too long")
+    return message
+
+
 def compensation_key(release: str) -> str:
     return f"release_compensation:{release}"
 
@@ -157,6 +177,23 @@ def release_slug(release: str) -> str:
     return re.sub(r"[^a-zA-Z0-9]+", "-", release).strip("-").lower()
 
 
+def release_groups(database_path: Path) -> list[str]:
+    connection = sqlite3.connect(f"file:{database_path}?mode=ro", uri=True)
+    try:
+        if connection.execute("PRAGMA quick_check").fetchone()[0] != "ok":
+            raise ReleaseUpdateError("Edge SQLite integrity check failed")
+        rows = connection.execute(
+            """
+            SELECT DISTINCT group_fingerprint
+            FROM player_accounts
+            ORDER BY group_fingerprint
+            """
+        ).fetchall()
+    finally:
+        connection.close()
+    return [str(row[0]) for row in rows]
+
+
 def checked_marker(raw: str, release: str) -> dict[str, Any]:
     expected_primogems = primogem_compensation_amount(release)
     try:
@@ -566,6 +603,34 @@ def run_release_update(
     }
 
 
+def run_release_lifecycle(
+    *,
+    release_file: Path,
+    database_path: Path,
+    delivery_helper: Path,
+    stage: str,
+    deliverer: Callable[[Path, str, str, str], None] = run_delivery_helper,
+) -> dict[str, Any]:
+    if stage not in LIFECYCLE_STAGES:
+        raise ReleaseUpdateError(f"unsupported lifecycle stage: {stage}")
+    release = read_release(release_file)
+    message = build_lifecycle_message(release, stage)
+    groups = release_groups(database_path)
+    delivered = 0
+    for group in groups:
+        group_hash = hashlib.sha256(group.encode()).hexdigest()[:12]
+        event_id = f"release-{release_slug(release)}-{stage}-{group_hash}"
+        deliverer(delivery_helper, group, message, event_id)
+        delivered += 1
+    return {
+        "status": "ok",
+        "release": release,
+        "stage": stage,
+        "groups": len(groups),
+        "delivered": delivered,
+    }
+
+
 def parse_args() -> argparse.Namespace:
     home = Path.home()
     parser = argparse.ArgumentParser(description=__doc__)
@@ -599,20 +664,33 @@ def parse_args() -> argparse.Namespace:
         action="store_true",
         help="retry only after visually confirming that the prior send did not land",
     )
+    parser.add_argument(
+        "--lifecycle-stage",
+        choices=sorted(LIFECYCLE_STAGES),
+        help="send a guarded lifecycle notice instead of release settlement",
+    )
     return parser.parse_args()
 
 
 def main() -> int:
     args = parse_args()
     try:
-        result = run_release_update(
+        if args.lifecycle_stage:
-            release_file=args.release_file,
+            result = run_release_lifecycle(
-            changelog=args.changelog,
+                release_file=args.release_file,
-            database_path=args.database,
+                database_path=args.database,
-            backup_dir=args.backup_dir,
+                delivery_helper=args.delivery_helper,
-            delivery_helper=args.delivery_helper,
+                stage=args.lifecycle_stage,
-            retry_uncertain=args.retry_uncertain_announcement,
+            )
-        )
+        else:
+            result = run_release_update(
+                release_file=args.release_file,
+                changelog=args.changelog,
+                database_path=args.database,
+                backup_dir=args.backup_dir,
+                delivery_helper=args.delivery_helper,
+                retry_uncertain=args.retry_uncertain_announcement,
+            )
     except ReleaseUpdateError as error:
         print(
             json.dumps(

packages/jwebot-config/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "jwebot-config"
-version = "1.1.1b1+1016"
+version = "1.1.2b1+1019"
 description = "Profile-based configuration compiler for JWeBot"
 requires-python = ">=3.9"
 dependencies = [

packages/jwebot-config/src/jwebot_config/__init__.py

@@ -36,4 +36,4 @@ __all__ = [
     "write_default_profile",
 ]
 
-__version__ = "1.1.1b1+1016"
+__version__ = "1.1.2b1+1019"

packages/jwebot-extension-host/pyproject.toml

@@ -4,12 +4,12 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "jwebot-extension-host"
-version = "1.1.1b1+1016"
+version = "1.1.2b1+1019"
 description = "Single-release, single-invocation process host for JWeBot extensions"
 requires-python = ">=3.9"
 dependencies = [
-  "jwebot-sdk==1.1.1b1+1016",
+  "jwebot-sdk==1.1.2b1+1019",
-  "jwebot-kernel==1.1.1b1+1016",
+  "jwebot-kernel==1.1.2b1+1019",
 ]
 
 [project.scripts]

packages/jwebot-extension-host/src/jwebot_extension_host/__init__.py

@@ -16,4 +16,4 @@ __all__ = [
     "process_request",
 ]
 
-__version__ = "1.1.1b1+1016"
+__version__ = "1.1.2b1+1019"

packages/jwebot-kernel/pyproject.toml

@@ -4,13 +4,13 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "jwebot-kernel"
-version = "1.1.1b1+1016"
+version = "1.1.2b1+1019"
 description = "Stable routing and extension authority for JWeBot"
 requires-python = ">=3.9"
 dependencies = [
-  "jwebot-config==1.1.1b1+1016",
+  "jwebot-config==1.1.2b1+1019",
-  "jwebot-sdk==1.1.1b1+1016",
+  "jwebot-sdk==1.1.2b1+1019",
-  "jwebot-official-extensions==1.1.1b1+1016",
+  "jwebot-official-extensions==1.1.2b1+1019",
   "tomli>=2; python_version < '3.11'",
 ]
 

packages/jwebot-kernel/src/jwebot_kernel/__init__.py

@@ -107,4 +107,4 @@ __all__ = [
     "verify_installed_release_signature",
 ]
 
-__version__ = "1.1.1b1+1016"
+__version__ = "1.1.2b1+1019"

packages/jwebot-kernel/src/jwebot_kernel/edge_runtime.py

@@ -23,6 +23,7 @@ import urllib.parse
 import urllib.request
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
+from functools import lru_cache
 from pathlib import Path
 from typing import Any, Callable, Iterable, Mapping
 from zoneinfo import ZoneInfo
@@ -132,6 +133,18 @@ CLOCK_GRACE_MINUTES = 3
 CLOCK_POLL_SECONDS = 15
 CLOCK_MIN_GROUP_MEMBERS = 2
 CLOCK_RANKING_LIMIT = 5
+AMBIENT_CHAT_START_HOUR = 10
+AMBIENT_CHAT_END_HOUR = 23
+AMBIENT_CHAT_ACTIVITY_MINUTES = 45
+AMBIENT_CHAT_MIN_CONTEXT_MESSAGES = 4
+AMBIENT_CHAT_MIN_CONTEXT_SPEAKERS = 2
+AMBIENT_CHAT_COOLDOWN_HOURS = 4
+AMBIENT_CHAT_PROBABILITY_PERCENT = 35
+TRIVIA_GAME_KEY = "trivia_quiz"
+TRIVIA_HISTORY_KEY = "trivia_quiz_history"
+TRIVIA_ANSWER_WINDOW_MINUTES = 5
+TRIVIA_QUIZ_PROBABILITY_PERCENT = 75
+TRIVIA_RECENT_BASE_LIMIT = 24
 BEIJING_TIMEZONE = ZoneInfo("Asia/Shanghai")
 ACKNOWLEDGEMENT = "已收到，正在处理中喵"
 STEAM_ACKNOWLEDGEMENT = "Steam 查询 [■□□] 正在读取公开数据"
@@ -195,9 +208,153 @@ TRIGGER = re.compile(
 PERSON_ALIAS_TEMPLATES: tuple[dict[str, object], ...] = ()
 PERSON_DISPLAY_IDS: dict[str, str] = {}
 CLOCK_SCAN_TARGETS: tuple[tuple[str, str], ...] = ()
+SCAN_BOOST_PITY_MISS_LIMIT = 2
 STEAM_GROUP_ROSTER: tuple[dict[str, object], ...] = ()
 PERSON_ALIAS_MIGRATION_VERSION = "person-aliases-empty"
 PROFILE_ALIAS_REPAIR_CANONICALS: frozenset[str] = frozenset()
+TRIVIA_QUESTIONS: tuple[dict[str, object], ...] = (
+    {
+        "id": "cs2-smoke-he",
+        "domain": "CS",
+        "question": "CS2 里哪种投掷物可以把烟雾短暂炸开一个缺口？",
+        "options": ("燃烧弹", "高爆手雷", "诱饵弹", "闪光弹"),
+        "answer": "B",
+        "explanation": "CS2 的体积烟会被 HE 爆炸短暂驱散，子弹也能打出小洞。",
+    },
+    {
+        "id": "csgo-danger-zone",
+        "domain": "CS",
+        "question": "CS:GO 最早的 Danger Zone 大逃杀地图叫什么？",
+        "options": ("Blacksite", "Basalt", "Breach", "Anubis"),
+        "answer": "A",
+        "explanation": "Danger Zone 随 Blacksite 首发，后来才有 Sirocco 等地图。",
+    },
+    {
+        "id": "apex-ash-name",
+        "domain": "Apex",
+        "question": "Apex 里 Ash 变成 Simulacrum 之前的人类名字是？",
+        "options": ("Amelie Paquette", "Ashleigh Reid", "Mary Somers", "Anita Williams"),
+        "answer": "B",
+        "explanation": "Ash 的人类身份是 Dr. Ashleigh Reid，和 Horizon 的过去有关。",
+    },
+    {
+        "id": "apex-horizon-son",
+        "domain": "Apex",
+        "question": "Apex 里 Horizon 一直想回去见到的儿子叫什么？",
+        "options": ("Newton", "Elliott", "Jackson", "D.O.C."),
+        "answer": "A",
+        "explanation": "Horizon 的儿子叫 Newton，名字也呼应了她的科学家设定。",
+    },
+    {
+        "id": "overwatch-sigma-name",
+        "domain": "Overwatch",
+        "question": "Overwatch 里 Sigma 的本名是？",
+        "options": ("Jamison Fawkes", "Siebren de Kuiper", "Tekhartha Mondatta", "Akande Ogundimu"),
+        "answer": "B",
+        "explanation": "Sigma 的本名是 Siebren de Kuiper，事故前是天体物理学家。",
+    },
+    {
+        "id": "overwatch-zen-title",
+        "domain": "Overwatch",
+        "question": "Zenyatta 名字前常见的 Tekhartha 更接近什么？",
+        "options": ("家族姓氏", "僧侣称号", "武器型号", "出生地名"),
+        "answer": "B",
+        "explanation": "Tekhartha 是香巴里僧侣相关称号，不是普通意义的姓。",
+    },
+    {
+        "id": "lol-ivern-cruel",
+        "domain": "LoL",
+        "question": "LoL 里 Ivern 成为绿树朋友前曾被称为什么？",
+        "options": ("Ivern the Cruel", "Ivern the Lost", "Ivern the Silent", "Ivern the First"),
+        "answer": "A",
+        "explanation": "Ivern 过去是 Ivern the Cruel，后来在神树事件后完全转变。",
+    },
+    {
+        "id": "lol-aurelion-crown",
+        "domain": "LoL",
+        "question": "Aurelion Sol 被巨神族控制主要靠什么东西束缚？",
+        "options": ("星界锁链", "一顶冠冕", "暮光长矛", "世界符文"),
+        "answer": "B",
+        "explanation": "巨神族用冠冕诱骗并束缚了 Aurelion Sol，这也是他怨气来源之一。",
+    },
+)
+TRIVIA_FACTS: tuple[tuple[str, str, str, str, tuple[str, str, str], str, int], ...] = (
+    ("cs-zeus", "CS", "哪件装备是一次性近距离电击武器？", "Zeus x27", ("P250", "XM1014", "Nova"), "Zeus x27 命中后通常只能打一发，是 CS 里很有梗的电击武器。", 3),
+    ("cs-defuse-kit", "CS", "哪件道具能缩短 CT 拆包时间？", "Defuse Kit", ("Kevlar", "Rescue Kit", "Tactical Shield"), "拆弹钳能把拆包时间从 10 秒缩短到 5 秒。", 2),
+    ("cs-blacksite", "CS", "CS:GO Danger Zone 首发地图是哪张？", "Blacksite", ("Sirocco", "Ember", "Basalt"), "Blacksite 是 Danger Zone 上线时的首发地图。", 4),
+    ("cs-he-smoke", "CS", "CS2 里能短暂驱散体积烟的爆炸物是？", "高爆手雷", ("燃烧弹", "诱饵弹", "闪光弹"), "CS2 体积烟会被 HE 爆炸短暂炸开缺口。", 4),
+    ("cs-subtick", "CS", "CS2 官方重点宣传的输入结算机制叫？", "Sub-tick", ("SourceTV", "Panorama", "Prime"), "CS2 用 Sub-tick 记录动作发生时刻，和传统 tick 观感不同。", 3),
+    ("cs-office", "CS", "Office 在 CS 系列里属于哪类经典模式地图？", "人质救援", ("爆破", "军备竞赛", "死亡竞赛"), "Office 是经典 hostage rescue 地图，不是拆包图。", 2),
+    ("cs-cache", "CS", "Cache 的主题环境最接近哪个区域？", "切尔诺贝利周边", ("威尼斯水城", "摩洛哥集市", "月球基地"), "Cache 的视觉主题来自切尔诺贝利/普里皮亚季工业区域。", 4),
+    ("apex-ash", "Apex", "Ash 成为 Simulacrum 前的人类名字是？", "Ashleigh Reid", ("Amelie Paquette", "Mary Somers", "Anita Williams"), "Ash 的旧身份是 Dr. Ashleigh Reid。", 4),
+    ("apex-horizon", "Apex", "Horizon 一直想回去见到的儿子叫什么？", "Newton", ("Elliott", "Jackson", "Crypto"), "Newton 是 Horizon 的儿子。", 3),
+    ("apex-crypto", "Apex", "Crypto 的本名是？", "Tae Joon Park", ("Octavio Silva", "Walter Fitzroy", "Makoa Gibraltar"), "Crypto 的本名是 Tae Joon Park。", 3),
+    ("apex-caustic", "Apex", "Caustic 的本名是？", "Alexander Nox", ("Elliott Witt", "Obi Edolasim", "Ajay Che"), "Caustic 是 Alexander Nox，背景里是毒气科学家。", 3),
+    ("apex-wattson", "Apex", "Wattson 的本名是？", "Natalie Paquette", ("Loba Andrade", "Mary Somers", "Anita Williams"), "Wattson 是 Natalie Paquette，和竞技场电力系统有关。", 3),
+    ("apex-fuse", "Apex", "Fuse 的本名是？", "Walter Fitzroy", ("Kaleb Cross", "Elliott Witt", "Tae Joon Park"), "Fuse 的本名是 Walter Fitzroy。", 3),
+    ("apex-seer", "Apex", "Seer 的本名是？", "Obi Edolasim", ("Makoa Gibraltar", "Octavio Silva", "Jackson Williams"), "Seer 的本名是 Obi Edolasim。", 4),
+    ("overwatch-sigma", "Overwatch", "Sigma 的本名是？", "Siebren de Kuiper", ("Akande Ogundimu", "Jamison Fawkes", "Mako Rutledge"), "Sigma 是 Siebren de Kuiper，事故前是天体物理学家。", 3),
+    ("overwatch-doomfist", "Overwatch", "Doomfist 的本名是？", "Akande Ogundimu", ("Tekhartha Zenyatta", "Gabriel Reyes", "Niran Pruksamanee"), "Doomfist 的本名是 Akande Ogundimu。", 3),
+    ("overwatch-junkrat", "Overwatch", "Junkrat 的本名是？", "Jamison Fawkes", ("Mako Rutledge", "Cole Cassidy", "Lúcio Correia dos Santos"), "Junkrat 的本名是 Jamison Fawkes。", 2),
+    ("overwatch-roadhog", "Overwatch", "Roadhog 的本名是？", "Mako Rutledge", ("Jamison Fawkes", "Baptiste Augustin", "Siebren de Kuiper"), "Roadhog 的本名是 Mako Rutledge。", 2),
+    ("overwatch-bastion", "Overwatch", "Bastion 的机体型号是？", "SST Laboratories Siege Automaton E54", ("MEKA Tokki", "R-7000 Ravager", "OR15 Defense Unit"), "Bastion 属于 E54 堡垒机器人。", 5),
+    ("overwatch-ramattra", "Overwatch", "Ramattra 领导的组织是？", "Null Sector", ("Talon", "Overwatch", "Vishkar"), "Ramattra 是 Null Sector 的核心领袖。", 3),
+    ("lol-ivern", "LoL", "Ivern 转变前曾被称为什么？", "Ivern the Cruel", ("Ivern the Lost", "Ivern the First", "Ivern the Quiet"), "Ivern 原本是残酷战士，后来因神树事件转变。", 4),
+    ("lol-aurelion", "LoL", "Aurelion Sol 被巨神族控制主要靠什么？", "一顶冠冕", ("世界符文", "暮光长矛", "暗裔剑"), "巨神族用冠冕束缚了 Aurelion Sol。", 4),
+    ("lol-swain", "LoL", "Swain 体内/身边关联的恶魔名字是？", "Raum", ("Tahm", "Evelynn", "Nocturne"), "Swain 与秘密恶魔 Raum 绑定。", 5),
+    ("lol-jhin", "LoL", "Jhin 的全名是？", "Khada Jhin", ("Shieda Kayn", "Jericho Swain", "Tobias Felix"), "Jhin 的名字是 Khada Jhin。", 3),
+    ("lol-rengar", "LoL", "Rengar 属于哪个瓦斯塔亚部族？", "Kiilash", ("Marai", "Lhotlan", "Oovi-Kat"), "Rengar 来自 Kiilash 部族。", 5),
+    ("lol-zac", "LoL", "Zac 名字常被解释为什么缩写？", "Zaun Amorphous Combatant", ("Zaun Arcane Core", "Z Drive Automaton", "Zaunite Acid Creature"), "Zac 是 Zaun Amorphous Combatant 的缩写梗。", 4),
+    ("lol-kindred", "LoL", "Kindred 由哪两个化身组成？", "Lamb 和 Wolf", ("Crow 和 Stag", "Fox 和 Owl", "Sun 和 Moon"), "Kindred 是 Lamb 与 Wolf 两个死亡化身。", 2),
+    ("lol-ryze", "LoL", "Ryze 长期收集并封存的危险物是什么？", "世界符文", ("暗裔武器", "海克斯核心", "虚空碎片"), "Ryze 的主线目标是阻止世界符文失控。", 3),
+)
+TRIVIA_TEMPLATES = (
+    "【{domain} 冷知识】{prompt}",
+    "Taffy 冷门题：{prompt}",
+    "不查资料猜一下：{prompt}",
+    "这个有点偏：{prompt}",
+    "冷知识选项题：{prompt}",
+    "老玩家小测：{prompt}",
+    "边角料问题：{prompt}",
+    "五分钟抢答：{prompt}",
+)
+TRIVIA_OPTION_LABELS = ("A", "B", "C", "D")
+
+
+def trivia_reward(rarity: int) -> int:
+    return 20 + max(1, min(5, int(rarity))) * 30
+
+
+@lru_cache(maxsize=1)
+def trivia_question_bank() -> tuple[dict[str, object], ...]:
+    questions: list[dict[str, object]] = [
+        {
+            **question,
+            "base_id": str(question["id"]),
+            "rarity": int(question.get("rarity", 4)),
+            "reward": trivia_reward(int(question.get("rarity", 4))),
+        }
+        for question in TRIVIA_QUESTIONS
+    ]
+    for base_id, domain, prompt, answer, distractors, explanation, rarity in TRIVIA_FACTS:
+        choices = (answer, *distractors)
+        for variant, template in enumerate(TRIVIA_TEMPLATES):
+            rotated = choices[variant % len(choices) :] + choices[: variant % len(choices)]
+            answer_label = TRIVIA_OPTION_LABELS[rotated.index(answer)]
+            questions.append(
+                {
+                    "id": f"{base_id}-{variant + 1}",
+                    "base_id": base_id,
+                    "domain": domain,
+                    "question": template.format(domain=domain, prompt=prompt),
+                    "options": rotated,
+                    "answer": answer_label,
+                    "explanation": explanation,
+                    "rarity": rarity,
+                    "reward": trivia_reward(rarity),
+                }
+            )
+    return tuple(questions)
 ALIAS_IDENTITY_QUESTION = re.compile(
     r"(?:谁(?:啊|呀)?|哪(?:位|个|个人)|什么人|叫(?:什么|啥)|全名)",
     re.IGNORECASE,
@@ -2799,6 +2956,55 @@ class Ledger:
             ).fetchone()
         return str(row["value"]) if row else ""
 
+    def recent_clock_broadcast(
+        self,
+        group_fingerprint: str,
+        *,
+        kind: str,
+        since: datetime,
+    ) -> dict[str, Any] | None:
+        group = bounded_text(group_fingerprint, limit=256)
+        if not group:
+            return None
+        cutoff = since.astimezone(timezone.utc).isoformat(timespec="milliseconds")
+        with self.lock:
+            row = self.connection.execute(
+                """
+                SELECT group_fingerprint, slot_key, local_date, kind, state,
+                       updated_at
+                FROM clock_broadcasts
+                WHERE group_fingerprint = ?
+                  AND kind = ?
+                  AND state IN ('pending', 'delivered')
+                  AND updated_at >= ?
+                ORDER BY updated_at DESC, slot_key DESC
+                LIMIT 1
+                """,
+                (group, kind, cutoff),
+            ).fetchone()
+        return dict(row) if row else None
+
+    def clock_broadcast_slot(
+        self,
+        group_fingerprint: str,
+        slot_key: str,
+    ) -> dict[str, Any] | None:
+        group = bounded_text(group_fingerprint, limit=256)
+        slot = bounded_text(slot_key, limit=256)
+        if not group or not slot:
+            return None
+        with self.lock:
+            row = self.connection.execute(
+                """
+                SELECT group_fingerprint, slot_key, local_date, kind, state,
+                       message
+                FROM clock_broadcasts
+                WHERE group_fingerprint = ? AND slot_key = ?
+                """,
+                (group, slot),
+            ).fetchone()
+        return dict(row) if row else None
+
     def memory_events(self, limit: int = 200) -> list[dict[str, Any]]:
         with self.lock:
             rows = self.connection.execute(
@@ -2876,8 +3082,48 @@ class Ledger:
                         slot_key,
                     ),
                 )
+            elif cursor.rowcount == 0:
+                row = self.connection.execute(
+                    """
+                    SELECT state
+                    FROM clock_broadcasts
+                    WHERE group_fingerprint = ? AND slot_key = ?
+                      AND kind = ? AND state = 'pending'
+                    """,
+                    (group_fingerprint, slot_key, kind),
+                ).fetchone()
+                if row is not None:
+                    return True
         return cursor.rowcount == 1
 
+    def due_pending_clock_broadcast(
+        self,
+        group_fingerprint: str,
+        local_now: datetime,
+    ) -> dict[str, Any] | None:
+        group = bounded_text(group_fingerprint, limit=256)
+        if not group:
+            return None
+        slot_keys = [
+            f"hourly:{local_now.date().isoformat()}T{local_now.hour:02d}",
+            f"daily:{local_now.date().isoformat()}",
+        ]
+        with self.lock:
+            row = self.connection.execute(
+                """
+                SELECT group_fingerprint, slot_key, local_date, kind, message,
+                       scan_person_key, scan_label
+                FROM clock_broadcasts
+                WHERE group_fingerprint = ?
+                  AND slot_key IN (?, ?)
+                  AND state = 'pending'
+                ORDER BY CASE kind WHEN 'hourly' THEN 0 ELSE 1 END, slot_key
+                LIMIT 1
+                """,
+                (group, slot_keys[0], slot_keys[1]),
+            ).fetchone()
+        return dict(row) if row else None
+
     def due_scheduled_taffy_speech(
         self,
         group_fingerprint: str,
@@ -3064,6 +3310,9 @@ class Ledger:
             )
             entry["weight"] += int(boost.get("extra_weight", 0))
         weight_list = list(weights.values())
+        total_weight = sum(int(item["weight"]) for item in weight_list) or 1
+        for item in weight_list:
+            item["share"] = int(item["weight"]) * 100 / total_weight
         weight_list.sort(
             key=lambda item: (
                 -int(item["weight"]),
@@ -3410,6 +3659,109 @@ class Ledger:
             (group, event_id, person_key, command, response, timestamp),
         )
 
+    def answer_trivia(
+        self,
+        group: str,
+        sender: str,
+        event_id: str,
+        answer: str,
+        *,
+        now: datetime | None = None,
+    ) -> str:
+        normalized_answer = normalize_text(answer).strip().upper()[:1]
+        if normalized_answer not in TRIVIA_OPTION_LABELS:
+            return "用法：/回答 A、/回答 B、/回答 C 或 /回答 D"
+        timestamp = utc_now()
+        current = now or datetime.now(timezone.utc)
+        if current.tzinfo is None:
+            current = current.replace(tzinfo=timezone.utc)
+        current_ts = current.timestamp()
+        command = f"trivia_answer:{normalized_answer}"
+        person = self._player_identity(group, sender)
+        person_key = str(person["person_key"])
+        display_name = str(person["display_name"])
+        with self.lock, self.connection:
+            replay = self._player_action_replay(group, event_id, person_key, command)
+            if replay is not None:
+                return replay
+            stored = self.connection.execute(
+                """
+                SELECT state_json FROM local_games
+                WHERE group_fingerprint = ? AND game_key = ?
+                """,
+                (group, TRIVIA_GAME_KEY),
+            ).fetchone()
+            if stored is None:
+                response = "当前没有正在进行的冷门问答，等 Taffy 下一题喵"
+                self._store_player_action(
+                    group, event_id, person_key, command, response, timestamp,
+                )
+                return response
+            try:
+                state = json.loads(str(stored["state_json"]))
+            except json.JSONDecodeError as error:
+                raise LocalCommandError("冷门问答状态无效") from error
+            if not isinstance(state, dict):
+                raise LocalCommandError("冷门问答状态无效")
+            try:
+                expires_at = float(state.get("expires_at", 0))
+            except (TypeError, ValueError):
+                expires_at = 0.0
+            correct_answer = str(state.get("answer", "")).upper()
+            explanation = bounded_text(state.get("explanation"), limit=180)
+            if current_ts > expires_at:
+                self.connection.execute(
+                    """
+                    DELETE FROM local_games
+                    WHERE group_fingerprint = ? AND game_key = ?
+                    """,
+                    (group, TRIVIA_GAME_KEY),
+                )
+                response = (
+                    f"这题已经超时了，正确答案是 {correct_answer}。"
+                    f"{explanation}"
+                )
+                self._store_player_action(
+                    group, event_id, person_key, command, response, timestamp,
+                )
+                return response
+            if normalized_answer != correct_answer:
+                response = (
+                    f"{display_name}：不对喵，5分钟内还能继续猜，"
+                    "用 /回答 A/B/C/D。"
+                )
+                self._store_player_action(
+                    group, event_id, person_key, command, response, timestamp,
+                )
+                return response
+            reward = max(1, min(500, int(state.get("reward", 50))))
+            account = self._load_player_account(group, person_key)
+            account["balance"] += reward
+            if account["balance"] > 1_000_000_000:
+                raise LocalCommandError("摩拉结算超出允许范围")
+            append_coin_event(
+                account,
+                reward,
+                f"冷门问答答对：{state.get('question_id', '')}",
+            )
+            self._save_player_account(group, person_key, account, timestamp)
+            self.connection.execute(
+                """
+                DELETE FROM local_games
+                WHERE group_fingerprint = ? AND game_key = ?
+                """,
+                (group, TRIVIA_GAME_KEY),
+            )
+            response = (
+                f"{display_name} 答对了：{correct_answer}。"
+                f"奖励摩拉 +{reward}，当前摩拉 {account['balance']}。"
+                f"{explanation}"
+            )
+            self._store_player_action(
+                group, event_id, person_key, command, response, timestamp,
+            )
+            return response
+
     def _open_group_lottery(self, group: str) -> sqlite3.Row | None:
         return self.connection.execute(
             """
@@ -5231,17 +5583,138 @@ class Ledger:
             normalize_player_account(account)
             for boost in account["scan_boosts"]:
                 target_key = str(boost["target_person_key"])
+                created_at = float(boost.get("created_at", 0) or 0)
                 current = boosts.setdefault(
                     target_key,
                     {
                         "person_key": target_key,
                         "display_name": str(boost["target_display_name"]),
                         "extra_weight": 0,
+                        "active_boosts": 0,
+                        "first_boosted_at": created_at,
                     },
                 )
                 current["extra_weight"] += int(boost["weight"])
+                current["active_boosts"] += 1
+                current["first_boosted_at"] = min(
+                    float(current["first_boosted_at"]),
+                    created_at,
+                )
         return boosts
 
+    @staticmethod
+    def _clock_slot_started_at(slot_key: str) -> float:
+        if not slot_key.startswith("hourly:"):
+            return 0.0
+        try:
+            local_time = datetime.strptime(
+                slot_key.removeprefix("hourly:"),
+                "%Y-%m-%dT%H",
+            ).replace(tzinfo=BEIJING_TIMEZONE)
+        except ValueError:
+            return 0.0
+        return local_time.timestamp()
+
+    def scan_boost_pity_status(
+        self,
+        group: str,
+        *,
+        local_now: datetime | None = None,
+    ) -> list[dict[str, Any]]:
+        group = bounded_text(group, limit=256)
+        boosts = self.active_scan_boost_targets(group)
+        if not group or not boosts:
+            return []
+        if local_now is None:
+            local_now = datetime.now(BEIJING_TIMEZONE)
+        if local_now.tzinfo is None:
+            local_now = local_now.replace(tzinfo=BEIJING_TIMEZONE)
+        local_now = local_now.astimezone(BEIJING_TIMEZONE)
+        current_slot_started_at = (
+            local_now.replace(minute=0, second=0, microsecond=0).timestamp()
+        )
+        earliest_boost = min(
+            float(boost.get("first_boosted_at", 0) or 0)
+            for boost in boosts.values()
+        )
+        with self.lock:
+            rows = self.connection.execute(
+                """
+                SELECT slot_key, scan_person_key, scan_label
+                FROM clock_broadcasts
+                WHERE group_fingerprint = ?
+                  AND kind = 'hourly'
+                  AND state = 'delivered'
+                ORDER BY slot_key
+                """,
+                (group,),
+            ).fetchall()
+        delivered: list[dict[str, Any]] = []
+        for row in rows:
+            slot_started_at = self._clock_slot_started_at(str(row["slot_key"]))
+            if (
+                slot_started_at
+                and slot_started_at >= earliest_boost
+                and slot_started_at < current_slot_started_at
+            ):
+                delivered.append(
+                    {
+                        "slot_started_at": slot_started_at,
+                        "scan_person_key": str(row["scan_person_key"]),
+                        "scan_label": str(row["scan_label"]),
+                    }
+                )
+        statuses: list[dict[str, Any]] = []
+        for target_key, boost in boosts.items():
+            first_boosted_at = float(boost.get("first_boosted_at", 0) or 0)
+            misses = 0
+            last_hit_at = 0.0
+            for row in delivered:
+                if row["slot_started_at"] < first_boosted_at:
+                    continue
+                if (
+                    row["scan_person_key"] == target_key
+                    or row["scan_label"] == str(boost["display_name"])
+                ):
+                    misses = 0
+                    last_hit_at = float(row["slot_started_at"])
+                else:
+                    misses += 1
+            statuses.append(
+                {
+                    "person_key": target_key,
+                    "display_name": str(boost["display_name"]),
+                    "extra_weight": int(boost.get("extra_weight", 0)),
+                    "active_boosts": int(boost.get("active_boosts", 0)),
+                    "first_boosted_at": first_boosted_at,
+                    "last_hit_at": last_hit_at,
+                    "misses": misses,
+                    "threshold": SCAN_BOOST_PITY_MISS_LIMIT,
+                    "guaranteed": misses >= SCAN_BOOST_PITY_MISS_LIMIT,
+                }
+            )
+        statuses.sort(
+            key=lambda item: (
+                not bool(item["guaranteed"]),
+                -int(item["misses"]),
+                -int(item["extra_weight"]),
+                float(item["first_boosted_at"]),
+                str(item["display_name"]).casefold(),
+            )
+        )
+        return statuses
+
+    def scan_boost_pity_candidate(
+        self,
+        group: str,
+        *,
+        local_now: datetime | None = None,
+    ) -> dict[str, Any] | None:
+        for status in self.scan_boost_pity_status(group, local_now=local_now):
+            if bool(status["guaranteed"]):
+                return status
+        return None
+
     def consume_scan_boost(
         self,
         group: str,
@@ -6013,6 +6486,220 @@ class HourlyBroadcastEngine:
             return now.replace(tzinfo=BEIJING_TIMEZONE)
         return now.astimezone(BEIJING_TIMEZONE)
 
+    @staticmethod
+    def _observed_at(value: object) -> datetime | None:
+        try:
+            observed = datetime.fromisoformat(str(value))
+        except (TypeError, ValueError):
+            return None
+        if observed.tzinfo is None:
+            observed = observed.replace(tzinfo=BEIJING_TIMEZONE)
+        return observed.astimezone(BEIJING_TIMEZONE)
+
+    @staticmethod
+    def _ambient_context_text(value: object) -> str:
+        text = normalize_text(value)
+        if not text or len(text) < 2:
+            return ""
+        if text.startswith("/") or f"@{BOT_MENTION}" in text or f"＠{BOT_MENTION}" in text:
+            return ""
+        if TOOL_REQUEST.search(text) or SENSITIVE_VALUE.search(text):
+            return ""
+        return bounded_text(text, limit=120)
+
+    def recent_ambient_context(
+        self,
+        group: str,
+        local_now: datetime,
+    ) -> list[dict[str, str]]:
+        cutoff = local_now - timedelta(minutes=AMBIENT_CHAT_ACTIVITY_MINUTES)
+        candidates: list[dict[str, str]] = []
+        try:
+            rows = self.ledger.recent_context_messages(group, 100)
+        except LocalCommandError:
+            return []
+        for row in rows:
+            observed = self._observed_at(row.get("observed_at"))
+            if observed is None or observed < cutoff or observed > local_now:
+                continue
+            text = self._ambient_context_text(row.get("text"))
+            if not text:
+                continue
+            display = bounded_text(row.get("display_name"), limit=40) or "群友"
+            candidates.append({"display_name": display, "text": text})
+        return candidates[-12:]
+
+    @staticmethod
+    def render_ambient_message(
+        context: list[dict[str, str]],
+        *,
+        seed: str,
+    ) -> str:
+        if not context:
+            return ""
+        joined = " ".join(item["text"] for item in context)
+        latest = context[-1]["text"]
+        snippet = re.sub(r"[\\r\\n]+", " ", latest).strip(" ，。！？!?")
+        snippet = bounded_text(snippet, limit=24)
+        if re.search(r"加班|上班|工作|开会|写代码|修 bug|修bug", joined, re.I):
+            return "我路过看一眼，你们这个加班味有点太浓了喵"
+        if re.search(r"困|睡|晚安|熬夜|起床", joined, re.I):
+            return "我冒个泡，困了就别硬撑，嘴硬也要休息喵"
+        if re.search(r"游戏|CS|开黑|Steam|自走棋|21点|彩票|原石|摩拉", joined, re.I):
+            return "刚刚这个游戏味我闻到了，先假装没看到你们又上头了喵"
+        templates = [
+            f"路过一下，你们刚刚聊到“{snippet}”，感觉这话题要开始发散了喵",
+            f"我冒个泡，刚刚这句“{snippet}”有点值得被记一笔喵",
+            "我路过看一眼，你们继续聊，我先假装自己很忙喵",
+        ]
+        return templates[deterministic_int(seed, 0, len(templates) - 1)]
+
+    def due_ambient_broadcast(
+        self,
+        *,
+        group: str,
+        local_now: datetime,
+    ) -> ClockBroadcast | None:
+        if not (AMBIENT_CHAT_START_HOUR <= local_now.hour <= AMBIENT_CHAT_END_HOUR):
+            return None
+        cooldown_since = local_now - timedelta(hours=AMBIENT_CHAT_COOLDOWN_HOURS)
+        if self.ledger.recent_clock_broadcast(
+            group,
+            kind="ambient_chat",
+            since=cooldown_since,
+        ):
+            return None
+        local_date = local_now.date().isoformat()
+        seed = f"{group}|{local_date}|{local_now.hour:02d}|ambient-chat"
+        if deterministic_int(seed, 0, 99) >= AMBIENT_CHAT_PROBABILITY_PERCENT:
+            return None
+        context = self.recent_ambient_context(group, local_now)
+        speakers = {item["display_name"] for item in context}
+        if (
+            len(context) < AMBIENT_CHAT_MIN_CONTEXT_MESSAGES
+            or len(speakers) < AMBIENT_CHAT_MIN_CONTEXT_SPEAKERS
+        ):
+            return None
+        message = self.render_ambient_message(context, seed=seed)
+        if not message:
+            return None
+        return ClockBroadcast(
+            group_fingerprint=group,
+            slot_key=f"hourly:{local_date}T{local_now.hour:02d}",
+            local_date=local_date,
+            kind="ambient_chat",
+            message=message,
+        )
+
+    def trivia_recent_base_ids(self, group: str) -> list[str]:
+        state = self.ledger.local_game(group, TRIVIA_HISTORY_KEY)
+        if not state:
+            return []
+        raw = state.get("recent_base_ids", [])
+        if not isinstance(raw, list):
+            return []
+        return [bounded_text(item, limit=80) for item in raw if item][-TRIVIA_RECENT_BASE_LIMIT:]
+
+    def remember_trivia_question(self, group: str, question: dict[str, object]) -> None:
+        base_id = bounded_text(question.get("base_id") or question.get("id"), limit=80)
+        if not base_id:
+            return
+        recent = [item for item in self.trivia_recent_base_ids(group) if item != base_id]
+        recent.append(base_id)
+        self.ledger.save_local_game(
+            group,
+            TRIVIA_HISTORY_KEY,
+            {"recent_base_ids": recent[-TRIVIA_RECENT_BASE_LIMIT:]},
+        )
+
+    def select_trivia_question(
+        self,
+        *,
+        group: str,
+        local_now: datetime,
+    ) -> dict[str, object]:
+        bank = trivia_question_bank()
+        recent = set(self.trivia_recent_base_ids(group))
+        seed = f"{group}|{local_now.date().isoformat()}|{local_now.hour:02d}|trivia"
+        start = deterministic_int(seed, 0, len(bank) - 1)
+        for offset in range(len(bank)):
+            candidate = bank[(start + offset) % len(bank)]
+            base_id = str(candidate.get("base_id") or candidate.get("id"))
+            if base_id not in recent:
+                return candidate
+        return bank[start]
+
+    @staticmethod
+    def render_trivia_message(question: dict[str, object]) -> str:
+        options = tuple(str(item) for item in question.get("options", ()))
+        if len(options) != 4:
+            raise LocalCommandError("冷门问答题库选项无效")
+        rarity = max(1, min(5, int(question.get("rarity", 3))))
+        reward = max(1, min(500, int(question.get("reward", trivia_reward(rarity)))))
+        lines = [
+            f"冷门知识问答（冷门度 {rarity}/5，限时5分钟，答对 +{reward} 摩拉）",
+            str(question.get("question", "")),
+        ]
+        lines.extend(
+            f"{label}. {option}"
+            for label, option in zip(TRIVIA_OPTION_LABELS, options)
+        )
+        lines.append("回答格式：/回答 A")
+        return "\n".join(lines)
+
+    def due_trivia_broadcast(
+        self,
+        *,
+        group: str,
+        local_now: datetime,
+    ) -> ClockBroadcast | None:
+        if not (AMBIENT_CHAT_START_HOUR <= local_now.hour <= AMBIENT_CHAT_END_HOUR):
+            return None
+        local_date = local_now.date().isoformat()
+        slot_key = f"hourly:{local_date}T{local_now.hour:02d}"
+        if self.ledger.clock_broadcast_slot(group, slot_key):
+            return None
+        active = self.ledger.local_game(group, TRIVIA_GAME_KEY)
+        if active:
+            try:
+                if float(active.get("expires_at", 0)) > local_now.timestamp():
+                    return None
+            except (TypeError, ValueError):
+                return None
+            self.ledger.clear_local_game(group, TRIVIA_GAME_KEY)
+        seed = f"{group}|{local_date}|{local_now.hour:02d}|trivia-probability"
+        if deterministic_int(seed, 0, 99) >= TRIVIA_QUIZ_PROBABILITY_PERCENT:
+            return None
+        question = self.select_trivia_question(group=group, local_now=local_now)
+        message = self.render_trivia_message(question)
+        expires_at = (
+            local_now + timedelta(minutes=TRIVIA_ANSWER_WINDOW_MINUTES)
+        ).timestamp()
+        self.ledger.save_local_game(
+            group,
+            TRIVIA_GAME_KEY,
+            {
+                "question_id": str(question["id"]),
+                "base_id": str(question.get("base_id") or question["id"]),
+                "question": str(question["question"]),
+                "options": list(question["options"]),
+                "answer": str(question["answer"]),
+                "explanation": str(question["explanation"]),
+                "rarity": int(question.get("rarity", 3)),
+                "reward": int(question.get("reward", 50)),
+                "created_at": local_now.timestamp(),
+                "expires_at": expires_at,
+            },
+        )
+        self.remember_trivia_question(group, question)
+        return ClockBroadcast(
+            group_fingerprint=group,
+            slot_key=slot_key,
+            local_date=local_date,
+            kind="trivia_quiz",
+            message=message,
+        )
+
     def due_broadcast(
         self,
         *,
@@ -6036,6 +6723,17 @@ class HourlyBroadcastEngine:
             )
         if not self.ledger.is_clock_broadcast_group(group):
             return None
+        pending = self.ledger.due_pending_clock_broadcast(group, local_now)
+        if pending is not None:
+            return ClockBroadcast(
+                group_fingerprint=str(pending["group_fingerprint"]),
+                slot_key=str(pending["slot_key"]),
+                local_date=str(pending["local_date"]),
+                kind=str(pending["kind"]),
+                message=str(pending["message"]),
+                scan_person_key=str(pending["scan_person_key"]),
+                scan_label=str(pending["scan_label"]),
+            )
         if local_now.hour == 0:
             previous_day = (local_now - timedelta(days=1)).date()
             local_date = previous_day.isoformat()
@@ -6060,6 +6758,14 @@ class HourlyBroadcastEngine:
                 message=message,
             )
 
+        trivia = self.due_trivia_broadcast(group=group, local_now=local_now)
+        if trivia is not None:
+            return trivia
+
+        ambient = self.due_ambient_broadcast(group=group, local_now=local_now)
+        if ambient is not None:
+            return ambient
+
         local_date = local_now.date().isoformat()
         weighted_targets: list[dict[str, str]] = [
             {
@@ -6081,12 +6787,22 @@ class HourlyBroadcastEngine:
             )
         if not weighted_targets:
             return None
-        target_index = deterministic_int(
+        pity_target = self.ledger.scan_boost_pity_candidate(
-            f"{group}|{local_date}|{local_now.hour:02d}|clock-scan",
+            group,
-            0,
+            local_now=local_now,
-            len(weighted_targets) - 1,
         )
-        target = weighted_targets[target_index]
+        if pity_target is None:
+            target_index = deterministic_int(
+                f"{group}|{local_date}|{local_now.hour:02d}|clock-scan",
+                0,
+                len(weighted_targets) - 1,
+            )
+            target = weighted_targets[target_index]
+        else:
+            target = {
+                "person_key": str(pity_target["person_key"]),
+                "label": str(pity_target["display_name"]),
+            }
         scan_label = target["label"]
         return ClockBroadcast(
             group_fingerprint=group,

packages/jwebot-official-extensions/pyproject.toml

@@ -4,11 +4,11 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "jwebot-official-extensions"
-version = "1.1.1b1+1016"
+version = "1.1.2b1+1019"
 description = "Reviewed local command adapters and broker implementation for JWeBot"
 requires-python = ">=3.9"
 dependencies = [
-  "jwebot-sdk==1.1.1b1+1016",
+  "jwebot-sdk==1.1.2b1+1019",
 ]
 
 [tool.setuptools.packages.find]

packages/jwebot-official-extensions/src/jwebot_official_extensions/__init__.py

@@ -90,4 +90,4 @@ __all__ = [
     "render_feature_list",
 ]
 
-__version__ = "1.1.1b1+1016"
+__version__ = "1.1.2b1+1019"

packages/jwebot-official-extensions/src/jwebot_official_extensions/adapters/trivia_quiz.py

@@ -0,0 +1,45 @@
+"""Generated reviewed adapter for trivia-quiz."""
+
+from jwebot_sdk import Event, ExtensionResult, ReplyPlan, SingleGroupContext
+
+
+PROFILE = 'trivia-quiz'
+AUTHOR = 'jwebot'
+DESCRIPTION = '冷门知识问答'
+HELP_ENTRIES = ('/回答 A；/answer A',)
+COMMANDS = ('/回答', '/answer')
+
+
+def _handler(context: SingleGroupContext, command: str):
+    def handle(event: Event, arguments: str) -> ExtensionResult:
+        response = context.services.call(
+            PROFILE,
+            operation="execute",
+            payload={
+                "command": command,
+                "arguments": arguments,
+                "event": event.to_mapping(),
+            },
+        )
+        texts = response.get("texts")
+        if isinstance(texts, list) and texts and all(
+            isinstance(item, str) and item.strip() for item in texts
+        ):
+            return ExtensionResult(
+                reply=ReplyPlan.multi_text(texts),
+                metadata={"local_extension": PROFILE, "author": AUTHOR},
+            )
+        text = response.get("text")
+        if not isinstance(text, str) or not text.strip():
+            raise ValueError("local service returned invalid text")
+        return ExtensionResult(
+            reply=ReplyPlan.text(text),
+            metadata={"local_extension": PROFILE, "author": AUTHOR},
+        )
+
+    return handle
+
+
+def plugin(context: SingleGroupContext) -> None:
+    for command in COMMANDS:
+        context.commands.register(command, _handler(context, command))

packages/jwebot-official-extensions/src/jwebot_official_extensions/descriptors.py

@@ -63,6 +63,8 @@ LOCAL_COMMAND_ALIASES = {
     "扫码概率": "scan_boost",
     "Taffy发话": "taffy_say",
     "taffysay": "taffy_say",
+    "回答": "trivia_answer",
+    "answer": "trivia_answer",
     "转账": "transfer",
     "transfer": "transfer",
     "贷款": "loan",
@@ -228,6 +230,13 @@ LOCAL_EXTENSIONS = (
         help_entries=("/贷款；/还款",),
         command_kinds=("loan", "repay"),
     ),
+    LocalExtensionDescriptor(
+        extension_id="trivia-quiz",
+        author=JWEBOT_AUTHOR,
+        description="冷门知识问答",
+        help_entries=("/回答 A；/answer A",),
+        command_kinds=("trivia_answer",),
+    ),
     LocalExtensionDescriptor(
         extension_id="cs-major-autochess",
         author=JWEBOT_AUTHOR,

packages/jwebot-official-extensions/src/jwebot_official_extensions/local_commands.py

@@ -222,6 +222,8 @@ class LocalCommandEngine:
             return self.reply_cleaner(self.deepseek_balance())
         if kind == "steam_friends":
             return self.reply_cleaner(self.steam_friends(arguments))
+        if kind == "trivia_answer":
+            return self.reply_cleaner(self.trivia_answer(row, arguments))
         raise LocalCommandError("unsupported local command")
 
     @staticmethod
@@ -261,10 +263,13 @@ class LocalCommandEngine:
             for index, entry in enumerate(ranking, start=1)
         )
         weight_listing = "；".join(
-            f"{entry['display_name']} {entry['weight']}"
+            f"{entry['display_name']} {entry['weight']}（{entry['share']:.1f}%）"
             for entry in weights
         )
-        return f"北京时间今日扫码排行：{listing}\n下一小时扫码权重：{weight_listing}"
+        return (
+            f"北京时间今日扫码排行：{listing}\n"
+            f"下一小时扫码权重：{weight_listing}"
+        )
 
     def repeat_ranking(self, row: dict[str, Any]) -> str:
         self.ledger.bootstrap_repeat_history()
@@ -620,6 +625,17 @@ class LocalCommandEngine:
             return "Steam 查询当前不可用"
         return self.steam.render(arguments)
 
+    def trivia_answer(self, row: dict[str, Any], arguments: str) -> str:
+        answer = normalize_text(arguments)
+        if not answer:
+            return "用法：/回答 A、/回答 B、/回答 C 或 /回答 D"
+        return self.ledger.answer_trivia(
+            str(row["group_fingerprint"]),
+            str(row["sender"]),
+            str(row["event_id"]),
+            answer,
+        )
+
     def profile_for_row(self, row: dict[str, Any]) -> dict[str, Any] | None:
         person = self.ledger.resolve_person(
             str(row["group_fingerprint"]),

packages/jwebot-official-extensions/src/jwebot_official_extensions/player_games.py

@@ -536,7 +536,7 @@ class PlayerGames:
             "同一时间最多一笔未还贷款。\n"
             "原石：达成特定成就获得；/原石祈愿 消耗1原石抽20～10000摩拉"
             "（理论期望3600摩拉）并可能获得扫码概率券；"
-            "/扫码加权 人名 提升TA的整点扫码权重；"
+            "/扫码加权 人名 在24小时内给TA整点扫码+5权重；"
             "/Taffy发话 内容 或 /Taffy发话 定时 HH:MM 内容。\n"
             "抽奖：/抽奖 [2～100]，5%返5倍、20%返2倍，其余无返还。\n"
             "多人彩票：/彩票 开始；/彩票 报名 3 4 7 [1～500]；"

packages/jwebot-sdk/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "jwebot-sdk"
-version = "1.1.1b1+1016"
+version = "1.1.2b1+1019"
 description = "Typed, capability-limited extension API for JWeBot plugins"
 requires-python = ">=3.9"
 dependencies = []

packages/jwebot-sdk/src/jwebot_sdk/__init__.py

@@ -38,4 +38,4 @@ __all__ = [
     "SingleGroupContext",
 ]
 
-__version__ = "1.1.1b1+1016"
+__version__ = "1.1.2b1+1019"

pyproject.toml

@@ -1,5 +1,5 @@
 [tool.jwebot]
-release = "beta 1.1.1.1016"
+release = "beta 1.1.2.1019"
 production_deployments_enabled_by_default = false
 
 [tool.jwebot.workspace]

tests/test_deployment_scripts.py

@@ -127,6 +127,10 @@ class DeploymentScriptTests(unittest.TestCase):
             'jwebot_log "runtime install ready"',
             'jwebot_log "edge active"',
             'jwebot_log "adaptive active"',
+            'send_release_lifecycle_notice shutdown',
+            'send_release_lifecycle_notice startup',
+            'jwebot_log "release lifecycle shutdown notice ready"',
+            'jwebot_log "release lifecycle startup notice ready"',
             'jwebot_log "release settlement starting"',
             '"${target_runtime}/jwebot_release_update.py"',
             'jwebot_log "release settlement ready"',
@@ -142,10 +146,20 @@ class DeploymentScriptTests(unittest.TestCase):
             source.count("xdotool"),
             source.count("/usr/bin/xdotool"),
         )
+        cutover_index = source.index('jwebot_log "cutover starting"')
+        actual_stop_index = source.index("user_systemctl stop", cutover_index)
+        self.assertLess(
+            source.index('jwebot_log "release lifecycle shutdown notice ready"'),
+            actual_stop_index,
+        )
         self.assertLess(
             source.index("trap - EXIT HUP INT TERM"),
             source.index('jwebot_log "release settlement starting"'),
         )
+        self.assertLess(
+            source.index('jwebot_log "release lifecycle startup notice ready"'),
+            source.index('jwebot_log "release settlement starting"'),
+        )
         self.assertLess(
             source.index('jwebot_log "release settlement ready"'),
             source.index('jwebot_log "deployment ready version='),

tests/test_jwebot_edge.py

@@ -8,6 +8,7 @@ import sqlite3
 import sys
 import tempfile
 import threading
+import time
 import unittest
 from pathlib import Path
 from unittest import mock
@@ -924,9 +925,9 @@ class LedgerAndApplicationTests(unittest.TestCase):
         self.assertEqual(version["route"], "local")
         self.assertIn("JWeBot 版本", version["text"])
         self.assertIn("kernel", version["text"])
-        self.assertIn("最新版本：beta 1.1.1.1016", version["text"])
+        self.assertIn("最新版本：beta 1.1.2.1019", version["text"])
-        self.assertIn("更新时间：2026-10-16 17:19:45", version["text"])
+        self.assertIn("更新时间：2026-10-19 16:38:10", version["text"])
-        self.assertIn("更新摘要：修复扫码加权重复购买", version["text"])
+        self.assertIn("更新摘要：新增低频主动插话、随机冷门知识问答", version["text"])
         self.assertEqual(debug["route"], "local")
         self.assertIn("调试埋点", debug["text"])
         self.assertEqual(balance["route"], "local")
@@ -1132,6 +1133,223 @@ class LedgerAndApplicationTests(unittest.TestCase):
         self.assertIn(ranking[0]["display_name"], daily_delivery["result_text"])
         self.application.deepseek.complete.assert_not_called()
 
+    def test_low_frequency_ambient_chat_uses_recent_context(self):
+        ledger = self.application.ledger
+        observed_at = MODULE.datetime(
+            2026,
+            10,
+            13,
+            12,
+            42,
+            tzinfo=MODULE.BEIJING_TIMEZONE,
+        ).isoformat()
+        for index, (sender, text) in enumerate(
+            (
+                ("Alice", "今天这个加班安排是不是有点离谱"),
+                ("Bob", "我也觉得还要继续修 bug"),
+                ("Alice", "先把问题处理完再说"),
+                ("Bob", "感觉咖啡已经救不了我了"),
+            ),
+            start=1,
+        ):
+            self.assertTrue(
+                ledger.record_person_observation(
+                    event_id=f"ambient-observation-{index}",
+                    group_fingerprint="group-a",
+                    sender=sender,
+                    text=text,
+                    observed_at=observed_at,
+                    source="test",
+                )
+            )
+            ledger.record_repeat_history(
+                event_id=f"ambient-context-{index}",
+                group_fingerprint="group-a",
+                sender=sender,
+                text=text,
+                observed_at=observed_at,
+            )
+        self.application.deliver = mock.Mock()
+        now = MODULE.datetime(
+            2026,
+            10,
+            13,
+            13,
+            0,
+            tzinfo=MODULE.BEIJING_TIMEZONE,
+        )
+
+        with mock.patch.object(MODULE, "TRIVIA_QUIZ_PROBABILITY_PERCENT", 0), \
+                mock.patch.object(MODULE, "AMBIENT_CHAT_PROBABILITY_PERCENT", 100):
+            self.assertTrue(
+                self.application.process_clock_once(
+                    now=now,
+                    group_fingerprint="group-a",
+                )
+            )
+
+        delivery = self.application.deliver.call_args.args[0]
+        self.assertIn("加班味", delivery["result_text"])
+        row = ledger.connection.execute(
+            """
+            SELECT kind, state FROM clock_broadcasts
+            WHERE group_fingerprint = ? AND slot_key = ?
+            """,
+            ("group-a", "hourly:2026-10-13T13"),
+        ).fetchone()
+        self.assertEqual(row["kind"], "ambient_chat")
+        self.assertEqual(row["state"], "delivered")
+        self.assertFalse(
+            self.application.process_clock_once(
+                now=now,
+                group_fingerprint="group-a",
+            )
+        )
+
+    def test_trivia_quiz_broadcast_and_answer_rewards_mora(self):
+        self.assertGreater(len(MODULE.trivia_question_bank()), 200)
+        ledger = self.application.ledger
+        observed_at = MODULE.datetime(
+            2026,
+            10,
+            13,
+            12,
+            42,
+            tzinfo=MODULE.BEIJING_TIMEZONE,
+        ).isoformat()
+        for sender in ("Alice", "Bob"):
+            ledger.record_person_observation(
+                event_id=f"trivia-member-{sender}",
+                group_fingerprint="group-a",
+                sender=sender,
+                text="冷门题准备一下",
+                observed_at=observed_at,
+                source="test",
+            )
+        self.application.deliver = mock.Mock()
+        now = MODULE.datetime(
+            2026,
+            10,
+            13,
+            13,
+            0,
+            tzinfo=MODULE.BEIJING_TIMEZONE,
+        )
+
+        with mock.patch.object(MODULE, "TRIVIA_QUIZ_PROBABILITY_PERCENT", 100):
+            self.assertTrue(
+                self.application.process_clock_once(
+                    now=now,
+                    group_fingerprint="group-a",
+                )
+            )
+
+        delivery = self.application.deliver.call_args.args[0]
+        self.assertIn("冷门知识问答", delivery["result_text"])
+        self.assertIn("限时5分钟", delivery["result_text"])
+        self.assertIn("/回答 A", delivery["result_text"])
+        state = ledger.local_game("group-a", MODULE.TRIVIA_GAME_KEY)
+        self.assertIsNotNone(state)
+        assert state is not None
+        state["expires_at"] = time.time() + 300
+        ledger.save_local_game("group-a", MODULE.TRIVIA_GAME_KEY, state)
+        answer = str(state["answer"])
+        reward = int(state["reward"])
+        result = self.application.accept(
+            event(
+                "trivia-answer",
+                f"@JWeBot /回答 {answer}",
+                sender="Alice",
+            )
+        )
+
+        self.assertEqual(result["route"], "local")
+        self.assertIn("答对了", result["text"])
+        self.assertIn(f"奖励摩拉 +{reward}", result["text"])
+        self.assertIsNone(ledger.local_game("group-a", MODULE.TRIVIA_GAME_KEY))
+        person = ledger.resolve_person("group-a", "Alice")
+        account = ledger._load_player_account("group-a", str(person["person_key"]))
+        self.assertEqual(account["balance"], 1000 + reward)
+
+    def test_ambient_chat_cooldown_keeps_next_hour_scan_broadcast(self):
+        ledger = self.application.ledger
+        observed_at = MODULE.datetime(
+            2026,
+            10,
+            13,
+            12,
+            45,
+            tzinfo=MODULE.BEIJING_TIMEZONE,
+        ).isoformat()
+        for index, (sender, text) in enumerate(
+            (
+                ("Alice", "今天加班真的有点顶不住"),
+                ("Bob", "先把这个线上问题看完"),
+                ("Alice", "咖啡续上"),
+                ("Bob", "修完再开一把游戏"),
+            ),
+            start=1,
+        ):
+            ledger.record_person_observation(
+                event_id=f"ambient-cooldown-observation-{index}",
+                group_fingerprint="group-a",
+                sender=sender,
+                text=text,
+                observed_at=observed_at,
+                source="test",
+            )
+            ledger.record_repeat_history(
+                event_id=f"ambient-cooldown-context-{index}",
+                group_fingerprint="group-a",
+                sender=sender,
+                text=text,
+                observed_at=observed_at,
+            )
+        self.application.deliver = mock.Mock()
+        first = MODULE.datetime(
+            2026,
+            10,
+            13,
+            13,
+            0,
+            tzinfo=MODULE.BEIJING_TIMEZONE,
+        )
+        second = MODULE.datetime(
+            2026,
+            10,
+            13,
+            14,
+            0,
+            tzinfo=MODULE.BEIJING_TIMEZONE,
+        )
+
+        with mock.patch.object(MODULE, "TRIVIA_QUIZ_PROBABILITY_PERCENT", 0), \
+                mock.patch.object(MODULE, "AMBIENT_CHAT_PROBABILITY_PERCENT", 100):
+            self.assertTrue(
+                self.application.process_clock_once(
+                    now=first,
+                    group_fingerprint="group-a",
+                )
+            )
+            self.assertTrue(
+                self.application.process_clock_once(
+                    now=second,
+                    group_fingerprint="group-a",
+                )
+            )
+
+        second_delivery = self.application.deliver.call_args.args[0]
+        self.assertIn("扫码了", second_delivery["result_text"])
+        row = ledger.connection.execute(
+            """
+            SELECT kind, state FROM clock_broadcasts
+            WHERE group_fingerprint = ? AND slot_key = ?
+            """,
+            ("group-a", "hourly:2026-10-13T14"),
+        ).fetchone()
+        self.assertEqual(row["kind"], "hourly")
+        self.assertEqual(row["state"], "delivered")
+
     def test_failed_hourly_broadcast_does_not_increment_scan_ranking(self):
         ledger = self.application.ledger
         for event_id, sender in (("clock-c", "Alice"), ("clock-d", "Bob")):
@@ -1162,6 +1380,134 @@ class LedgerAndApplicationTests(unittest.TestCase):
         self.assertEqual(ledger.clock_scan_ranking("group-a", "2026-10-13"), [])
         self.assertEqual(ledger.status()["clock_broadcasts_pending"], 0)
 
+    def test_existing_pending_hourly_broadcast_is_delivered(self):
+        ledger = self.application.ledger
+        for event_id, sender in (("clock-pending-a", "Alice"), ("clock-pending-b", "Bob")):
+            ledger.record_person_observation(
+                event_id=event_id,
+                group_fingerprint="group-a",
+                sender=sender,
+                text="群内聊天记录",
+                observed_at=MODULE.utc_now(),
+                source="test",
+            )
+        person = ledger.resolve_person("group-a", "Bob")
+        self.assertTrue(
+            ledger.reserve_clock_broadcast(
+                group_fingerprint="group-a",
+                slot_key="hourly:2026-10-13T05",
+                local_date="2026-10-13",
+                kind="hourly",
+                message="现在是北京时间 5:00，bob 扫码了。",
+                scan_person_key=str(person["person_key"]),
+                scan_label="bob",
+            )
+        )
+        self.application.deliver = mock.Mock()
+        now = MODULE.datetime(
+            2026,
+            10,
+            13,
+            5,
+            0,
+            tzinfo=MODULE.BEIJING_TIMEZONE,
+        )
+
+        self.assertTrue(
+            self.application.process_clock_once(
+                now=now,
+                group_fingerprint="group-a",
+            )
+        )
+
+        delivery = self.application.deliver.call_args.args[0]
+        self.assertIn("bob 扫码了", delivery["result_text"])
+        self.assertEqual(
+            ledger.clock_scan_ranking("group-a", "2026-10-13")[0]["display_name"],
+            "bob",
+        )
+
+    def test_scan_boost_pity_forces_target_after_two_misses(self):
+        ledger = self.application.ledger
+        for event_id, sender in (("pity-member-a", "Alice"), ("pity-member-b", "Bob")):
+            ledger.record_person_observation(
+                event_id=event_id,
+                group_fingerprint="group-a",
+                sender=sender,
+                text="群内聊天记录",
+                observed_at=MODULE.utc_now(),
+                source="test",
+            )
+        alice = ledger.resolve_person("group-a", "Alice")
+        bob = ledger.resolve_person("group-a", "Bob")
+        carol = ledger.resolve_person("group-a", "Carol")
+        account = MODULE.normalize_player_account({})
+        account["scan_boosts"] = [
+            {
+                "target_person_key": str(bob["person_key"]),
+                "target_display_name": "bob",
+                "weight": 5,
+                "created_at": MODULE.datetime(
+                    2026,
+                    10,
+                    13,
+                    3,
+                    30,
+                    tzinfo=MODULE.BEIJING_TIMEZONE,
+                ).timestamp(),
+                "expires_at": MODULE.datetime(
+                    2099,
+                    10,
+                    14,
+                    3,
+                    30,
+                    tzinfo=MODULE.BEIJING_TIMEZONE,
+                ).timestamp(),
+            }
+        ]
+        ledger._save_player_account(
+            "group-a",
+            str(alice["person_key"]),
+            account,
+            MODULE.utc_now(),
+        )
+        for hour, person, label in (
+            (4, alice, "alice"),
+            (5, carol, "carol"),
+        ):
+            slot_key = f"hourly:2026-10-13T{hour:02d}"
+            self.assertTrue(
+                ledger.reserve_clock_broadcast(
+                    group_fingerprint="group-a",
+                    slot_key=slot_key,
+                    local_date="2026-10-13",
+                    kind="hourly",
+                    message="test",
+                    scan_person_key=str(person["person_key"]),
+                    scan_label=label,
+                )
+            )
+            self.assertTrue(ledger.complete_clock_broadcast("group-a", slot_key))
+        now = MODULE.datetime(
+            2026,
+            10,
+            13,
+            6,
+            0,
+            tzinfo=MODULE.BEIJING_TIMEZONE,
+        )
+
+        status = ledger.scan_boost_pity_status("group-a", local_now=now)
+        self.assertTrue(status[0]["guaranteed"])
+        self.assertEqual(status[0]["display_name"], "bob")
+        broadcast = self.application.clock.due_broadcast(
+            now=now,
+            group_fingerprint="group-a",
+        )
+
+        self.assertIsNotNone(broadcast)
+        self.assertEqual(broadcast.scan_label, "bob")
+
     def test_scan_ranking_command_uses_current_beijing_date(self):
         ledger = self.application.ledger
         local_date = MODULE.datetime.now(MODULE.BEIJING_TIMEZONE).date().isoformat()

tests/test_official_extension_inventory.py

@@ -96,6 +96,7 @@ class LocalExtensionInventoryTests(unittest.TestCase):
                 "coin-taffy": JWEBOT_AUTHOR,
                 "primogem-economy": JWEBOT_AUTHOR,
                 "coin-loan": JWEBOT_AUTHOR,
+                "trivia-quiz": JWEBOT_AUTHOR,
                 "random-number": JWEBOT_AUTHOR,
                 "draw-lot": JWEBOT_AUTHOR,
                 "daily-fortune": JWEBOT_AUTHOR,

tests/test_release_update.py

@@ -20,7 +20,7 @@ SPEC.loader.exec_module(MODULE)
 
 
 class ReleaseUpdateTests(unittest.TestCase):
-    release = "beta 1.1.1.1016"
+    release = "beta 1.1.2.1019"
 
     def setUp(self):
         self.temporary = tempfile.TemporaryDirectory()
@@ -122,12 +122,34 @@ class ReleaseUpdateTests(unittest.TestCase):
             self.assertIn("更新补偿：摩拉 x200，原石 x1", announcement)
 
     def test_patch_release_without_policy_does_not_announce_primogem(self):
-        release = "beta 1.1.1.1016"
+        release = "beta 1.1.2.1019"
         self.assertEqual(MODULE.primogem_compensation_amount(release), 0)
         announcement = MODULE.build_announcement(release, "普通小版本更新。")
         self.assertIn("更新补偿：摩拉 x200", announcement)
         self.assertNotIn("原石 x1", announcement)
 
+    def test_lifecycle_stage_sends_guarded_notice_without_compensation(self):
+        delivered = []
+
+        def deliverer(path, group, message, event_id):
+            delivered.append((path, group, message, event_id))
+
+        result = MODULE.run_release_lifecycle(
+            release_file=self.version,
+            database_path=self.database,
+            delivery_helper=self.delivery_helper,
+            stage="shutdown",
+            deliverer=deliverer,
+        )
+
+        self.assertEqual(result["stage"], "shutdown")
+        self.assertEqual(result["delivered"], 1)
+        self.assertEqual(delivered[0][1], "group-a")
+        self.assertIn("准备停机更新", delivered[0][2])
+        self.assertIn(self.release, delivered[0][2])
+        self.assertIn("shutdown", delivered[0][3])
+        self.assertEqual([state["balance"] for state in self.account_states()], [1000, 50])
+
     def test_release_update_grants_once_and_announces_once(self):
         delivered = []