diff --git a/.github/workflows/pr-agent.yml b/.github/workflows/pr-agent.yml
index dae53fbf..63472d98 100644
--- a/.github/workflows/pr-agent.yml
+++ b/.github/workflows/pr-agent.yml
@@ -1,4 +1,4 @@
-name: PR Agent
+name: PR-Agent
 
 on:
   pull_request_target:
@@ -11,8 +11,7 @@ on:
       - review_requested
       - synchronize
   issue_comment:
-    # 手动命令如 "/describe"、"/improve" 和 "/ask ..." 只在 PR 评论中有意义。
-    # issue_comment 同时覆盖普通 issue，因此 job 里还会再判断是否属于 PR。
+    # 手动命令只在 PR 评论中有意义；编辑后的命令也可重新触发。
     types:
       - created
       - edited
@@ -22,16 +21,17 @@ permissions:
   contents: read
   # 更新 PR 描述、发布 PR Review 或修改 PR 相关元数据。
   pull-requests: write
-  # PR 评论在 GitHub API 中属于 issue comments，手动命令和总结评论需要该权限。
+  # PR 评论在 GitHub API 中属于 issue comments，手动问答需要该权限。
   issues: write
 
 jobs:
   pr-agent:
-    name: PR-Agent inline review
     if: >-
       github.event.sender.type != 'Bot' &&
       (
-        github.event_name == 'pull_request_target' ||
+        (
+          github.event_name == 'pull_request_target'
+        ) ||
         (
           github.event_name == 'issue_comment' &&
           github.event.issue.pull_request != null &&
@@ -39,545 +39,498 @@ jobs:
           (
             github.event.comment.body == '/describe' ||
             startsWith(github.event.comment.body, '/describe ') ||
-            github.event.comment.body == '/improve' ||
-            startsWith(github.event.comment.body, '/improve ') ||
+            github.event.comment.body == '/review' ||
+            startsWith(github.event.comment.body, '/review ') ||
             github.event.comment.body == '/ask' ||
             startsWith(github.event.comment.body, '/ask ')
           )
         )
       )
     concurrency:
-      group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.event.issue.number || github.ref }}
+      group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.event.issue.number }}
       cancel-in-progress: ${{ github.event_name == 'pull_request_target' }}
     runs-on: ubuntu-latest
     timeout-minutes: 20
+
     steps:
-      - name: Detect PR review language
-        id: pr_language
+      - name: Capture PR context
+        id: pr_context
         env:
           GH_TOKEN: ${{ github.token }}
           REPO: ${{ github.repository }}
           PR_NUMBER: ${{ github.event.pull_request.number || github.event.issue.number }}
         run: |
           set -euo pipefail
-          pr_info="$(mktemp)"
-          gh api "repos/${REPO}/pulls/${PR_NUMBER}" > "${pr_info}"
-          python3 - "${pr_info}" >> "${GITHUB_OUTPUT}" <<'PY'
+          pull="$(mktemp)"
+          gh api "repos/${REPO}/pulls/${PR_NUMBER}" > "${pull}"
+          python3 - "${pull}" >> "${GITHUB_OUTPUT}" <<'PY'
           import json
           import re
           import sys
           from pathlib import Path
 
-          pr = json.loads(Path(sys.argv[1]).read_text())
-          title = pr.get("title") or ""
-          body = pr.get("body") or ""
-          labels = {item.get("name", "") for item in pr.get("labels") or []}
+          pull = json.loads(Path(sys.argv[1]).read_text(encoding="utf-8"))
+          title = pull.get("title") or ""
+          body = pull.get("body") or ""
+          labels = {item.get("name", "") for item in pull.get("labels") or []}
           skip_pr_agent = "true" if "skip pr-agent" in labels or re.search(r"^(?:\[Auto\]|Auto)", title) else "false"
-          head_sha = pr.get("head", {}).get("sha") or ""
-
-          body = re.sub(
-              r"\n*##\s+(PR-Agent\s+摘要|PR-Agent Summary)\s*\n\s*"
-              r"<!-- pr-agent-summary:start -->.*?<!-- pr-agent-summary:end -->",
-              " ",
-              body,
-              flags=re.IGNORECASE | re.DOTALL,
-          )
           body = re.sub(
               r"<!-- pr-agent-summary:start -->.*?<!-- pr-agent-summary:end -->",
               " ",
               body,
               flags=re.DOTALL,
           )
-          body = re.sub(r"```.*?```", " ", body, flags=re.DOTALL)
-
           text = f"{title}\n{body}"
           cjk_count = len(re.findall(r"[\u4e00-\u9fff]", text))
           latin_words = len(re.findall(r"\b[A-Za-z][A-Za-z]{2,}\b", text))
-
-          if cjk_count >= 4:
+          if cjk_count >= 4 or latin_words < 8:
               response_language = "zh-CN"
               summary_heading = "PR-Agent 摘要"
-              summary_language = "中文"
-          elif latin_words >= 8:
+          else:
               response_language = "en-US"
               summary_heading = "PR-Agent Summary"
-              summary_language = "English"
-          else:
-              response_language = "zh-CN"
-              summary_heading = "PR-Agent 摘要"
-              summary_language = "中文"
-
+          print(f"head_sha={pull['head']['sha']}")
+          print(f"changed_files={pull.get('changed_files') or 0}")
           print(f"response_language={response_language}")
           print(f"summary_heading={summary_heading}")
-          print(f"summary_language={summary_language}")
           print(f"skip_pr_agent={skip_pr_agent}")
-          print(f"head_sha={head_sha}")
           PY
 
       - name: Prepare PR-Agent description markers
-        id: prepare_description
         if: >-
-          steps.pr_language.outputs.skip_pr_agent != 'true' &&
+          steps.pr_context.outputs.skip_pr_agent != 'true' &&
           (
             github.event_name == 'pull_request_target' ||
-            (
-              github.event_name == 'issue_comment' &&
-              (
-                github.event.comment.body == '/describe' ||
-                startsWith(github.event.comment.body, '/describe ')
-              )
-            )
+            github.event.comment.body == '/describe' ||
+            startsWith(github.event.comment.body, '/describe ')
           )
         env:
           GH_TOKEN: ${{ github.token }}
           REPO: ${{ github.repository }}
           PR_NUMBER: ${{ github.event.pull_request.number || github.event.issue.number }}
-          SUMMARY_HEADING: ${{ steps.pr_language.outputs.summary_heading }}
+          SUMMARY_HEADING: ${{ steps.pr_context.outputs.summary_heading }}
+          CHANGED_FILES: ${{ steps.pr_context.outputs.changed_files }}
         run: |
           set -euo pipefail
-          current_body="$(mktemp)"
-          next_body="$(mktemp)"
           payload="$(mktemp)"
-          body_backup="${RUNNER_TEMP}/pr-agent-body-before-describe.md"
-          placeholder_body="${RUNNER_TEMP}/pr-agent-body-with-placeholder.md"
-
-          gh api "repos/${REPO}/pulls/${PR_NUMBER}" --jq '.body // ""' > "${current_body}"
-          cp "${current_body}" "${body_backup}"
-          python3 - "${current_body}" "${next_body}" <<'PY'
+          python3 - "${payload}" <<'PY'
+          import json
           import os
           import re
+          import subprocess
           import sys
-          from pathlib import Path
 
-          current_path = Path(sys.argv[1])
-          next_path = Path(sys.argv[2])
-
-          body = current_path.read_text()
+          pull = json.loads(subprocess.check_output(
+              ["gh", "api", f"repos/{os.environ['REPO']}/pulls/{os.environ['PR_NUMBER']}"],
+              text=True,
+          ))
+          body = pull.get("body") or ""
           start = "<!-- pr-agent-summary:start -->"
           end = "<!-- pr-agent-summary:end -->"
-          placeholder = "pr_agent:summary"
-          summary_heading = os.environ.get("SUMMARY_HEADING") or "PR-Agent 摘要"
-          agent_block = f"## {summary_heading}\n\n{start}\n{placeholder}\n{end}\n"
-          body = re.sub(
-              r"(?im)^##\s+(PR-Agent\s+摘要|PR-Agent Summary)\s*\n\s*(?=<!-- pr-agent-summary:start -->)",
-              f"## {summary_heading}\n\n",
-              body,
-          )
-
+          heading = os.environ.get("SUMMARY_HEADING") or "PR-Agent 摘要"
+          changed_files = int(os.environ.get("CHANGED_FILES") or 0)
+          block = f"## {heading}\n\n{start}\npr_agent:summary\n{end}"
           start_index = body.find(start)
-          end_index = body.find(end)
-          if start_index >= 0 and end_index > start_index:
-              next_body = body[: start_index + len(start)] + f"\n{placeholder}\n" + body[end_index:]
-          else:
-              separator = "\n\n" if body.strip() else ""
-              next_body = body.rstrip() + separator + agent_block
-
-          next_path.write_text(next_body)
-          PY
-          cp "${next_body}" "${placeholder_body}"
-
-          body_changed=false
-          if ! cmp -s "${current_body}" "${next_body}"; then
-            python3 - "${next_body}" "${payload}" <<'PY'
-          import json
-          import sys
-          from pathlib import Path
-
-          body = Path(sys.argv[1]).read_text()
-          Path(sys.argv[2]).write_text(json.dumps({"body": body}, ensure_ascii=False))
-          PY
-            gh api --method PATCH "repos/${REPO}/pulls/${PR_NUMBER}" --input "${payload}" >/dev/null
-            body_changed=true
-          fi
-          echo "body_changed=${body_changed}" >> "${GITHUB_OUTPUT}"
-
-      - name: Snapshot PR-Agent inline comments
-        id: inline_state_before
-        if: >-
-          steps.pr_language.outputs.skip_pr_agent != 'true' &&
-          (
-            github.event_name == 'pull_request_target' ||
-            (
-              github.event_name == 'issue_comment' &&
-              (
-                github.event.comment.body == '/improve' ||
-                startsWith(github.event.comment.body, '/improve ')
-              )
-            )
+          end_index = body.find(end, start_index + len(start)) if start_index >= 0 else -1
+          owned_block = re.compile(
+              r"(?ims)^##\s+(?:PR-Agent\s+摘要|PR-Agent\s+Summary)\s*\n\s*"
+              r"<!-- pr-agent-summary:start -->.*?<!-- pr-agent-summary:end -->\s*"
           )
-        env:
-          GH_TOKEN: ${{ github.token }}
-          REPO: ${{ github.repository }}
-          PR_NUMBER: ${{ github.event.pull_request.number || github.event.issue.number }}
-        run: |
-          set -euo pipefail
-          inline_ids="$(gh api --paginate "repos/${REPO}/pulls/${PR_NUMBER}/comments?per_page=100" --jq '.[] | select(.user.login == "github-actions[bot]") | .id' | jq -sc '.')"
-          inline_ids_b64="$(printf '%s' "${inline_ids}" | base64 -w0)"
-          echo "inline_ids_b64=${inline_ids_b64}" >> "${GITHUB_OUTPUT}"
-
-      - name: Run PR-Agent
-        id: pragent
-        if: steps.pr_language.outputs.skip_pr_agent != 'true'
-        # 使用版本号加 digest 固定容器构建，避免 tag 被重推后改变运行内容。
-        uses: docker://pragent/pr-agent:0.39.0-github_action@sha256:b253845caa8c7ff5ce8be78f32996647982bdd4890826a962b78eff2e385a825
-        env:
-          # PR-Agent 使用该 token 读取 PR 元数据并发布评论。
-          GITHUB_TOKEN: ${{ github.token }}
-
-          # 仓库设置中添加的 Secret：Settings -> Secrets and variables -> Actions。
-          # 该 key 只传给 PR-Agent 运行时，不写入仓库。
-          OPENAI_KEY: ${{ secrets.OPENAI_KEY }}
-
-          # 仓库设置中添加的 Secret。OpenAI 兼容服务通常需要填写以 "/v1" 结尾的 API 根地址。
-          OPENAI.API_BASE: ${{ secrets.OPENAI_API_BASE }}
-
-          # 模型、输出语言和大 diff 处理策略。
-          config.model: "gpt-5.5"
-          config.fallback_models: '["gpt-5.4"]'
-          config.reasoning_effort: "xhigh"
-          config.ai_timeout: "900"
-          config.response_language: ${{ steps.pr_language.outputs.response_language }}
-          config.large_patch_policy: "clip"
-          config.ignore_pr_title: '["^\\[Auto\\]", "^Auto"]'
-          config.ignore_pr_labels: '["skip pr-agent"]'
-
-          # PR 初次进入评审或后续 push 时，更新 PR 摘要并发布 GitHub Review 行内建议。
-          github_action_config.auto_review: "false"
-          github_action_config.auto_describe: "true"
-          github_action_config.auto_improve: "true"
-
-          # synchronize 由 push_commands 单独处理；每次 push 更新摘要和行内 Review，旧行评由 GitHub 标记 outdated。
-          github_action_config.pr_actions: '["opened", "reopened", "ready_for_review", "review_requested"]'
-          github_action_config.handle_push_trigger: "true"
-          github_action_config.push_commands: '["/describe", "/improve"]'
-
-          # 保留 action outputs，便于后续 workflow 编排或排查。
-          github_action_config.enable_output: "true"
-
-          # /describe 行为控制；只更新 PR body 中的 PR-Agent 摘要占位符。
-          pr_description.generate_ai_title: "false"
-          pr_description.publish_labels: "false"
-          pr_description.publish_description_as_comment: "false"
-          pr_description.publish_description_as_comment_persistent: "false"
-          pr_description.enable_pr_diagram: "false"
-          pr_description.enable_pr_type: "false"
-          pr_description.enable_help_text: "false"
-          pr_description.enable_help_comment: "false"
-          pr_description.enable_semantic_files_types: "false"
-          pr_description.collapsible_file_list: "adaptive"
-          pr_description.add_original_user_description: "true"
-          pr_description.use_description_markers: "true"
-          pr_description.final_update_message: "false"
-          pr_description.extra_instructions: |
-            Match the configured response language.
-            Generate a moderately detailed PR summary covering the change goal, key implementation details, configuration or compatibility impact, tests, and notable risks.
-            Use 2-4 bullets for small PRs; use 4-8 bullets for feature or multi-file PRs.
-            Avoid low-value file lists and local command transcripts.
-
-          # /improve 以 GitHub 内联建议呈现，便于像正式 review discussion 一样逐条处理。
-          pr_code_suggestions.extra_instructions: |
-            Match the configured response language.
-            Only provide substantive issues that maintainers should address; avoid style-only, preference-only, or low-value suggestions.
-            For prioritized issues, start the suggestion body with one of these Markdown prefixes: 🔴 **High Risk**:, 🟡 **Medium Risk**:, or 🔵 **Low Risk**:.
-          pr_code_suggestions.focus_only_on_problems: "true"
-          pr_code_suggestions.suggestions_score_threshold: "3"
-          pr_code_suggestions.num_code_suggestions_per_chunk: "3"
-          pr_code_suggestions.commitable_code_suggestions: "true"
-          pr_code_suggestions.publish_output_no_suggestions: "false"
-          pr_questions.use_conversation_history: "true"
-
-          # 可选成本和噪音控制：
-          # github_action_config.auto_improve: "true"
-          # config.verbosity_level: "1"
-
-      - name: Publish PR-Agent code review summary
-        if: >-
-          steps.pr_language.outputs.skip_pr_agent != 'true' &&
-          (
-            github.event_name == 'pull_request_target' ||
-            (
-              github.event_name == 'issue_comment' &&
-              (
-                github.event.comment.body == '/improve' ||
-                startsWith(github.event.comment.body, '/improve ')
+          if changed_files == 0:
+              updated = owned_block.sub("", body)
+              if updated != body:
+                  updated = updated.rstrip()
+          elif start_index >= 0 and end_index >= 0:
+              body = re.sub(
+                  r"(?im)^##\s+(PR-Agent\s+摘要|PR-Agent\s+Summary)\s*\n\s*(?=<!-- pr-agent-summary:start -->)",
+                  f"## {heading}\n\n",
+                  body,
               )
-            )
-          )
-        env:
-          GH_TOKEN: ${{ github.token }}
-          REPO: ${{ github.repository }}
-          PR_NUMBER: ${{ github.event.pull_request.number || github.event.issue.number }}
-          BEFORE_INLINE_IDS_B64: ${{ steps.inline_state_before.outputs.inline_ids_b64 }}
-          RUN_HEAD_SHA: ${{ steps.pr_language.outputs.head_sha }}
-          SUMMARY_LANGUAGE: ${{ steps.pr_language.outputs.summary_language }}
-          OPENAI_KEY: ${{ secrets.OPENAI_KEY }}
-          OPENAI_API_BASE: ${{ secrets.OPENAI_API_BASE }}
-          SUMMARY_MODEL: gpt-5.5
-        run: |
-          set -euo pipefail
-          before_ids="$(printf '%s' "${BEFORE_INLINE_IDS_B64:-W10=}" | base64 -d)"
-          pr_info="$(mktemp)"
-          comments="$(mktemp)"
-          review_data="$(mktemp)"
-          payload="$(mktemp)"
-
-          gh api "repos/${REPO}/pulls/${PR_NUMBER}" > "${pr_info}"
-          CURRENT_HEAD_SHA="$(jq -r '.head.sha' "${pr_info}")"
-          HEAD_SHA="${RUN_HEAD_SHA:-${CURRENT_HEAD_SHA}}"
-          if [ "${CURRENT_HEAD_SHA}" != "${HEAD_SHA}" ]; then
-            echo "PR head changed from ${HEAD_SHA} to ${CURRENT_HEAD_SHA}; skip stale code review summary."
-            exit 0
-          fi
-          short_sha="${HEAD_SHA:0:7}"
-          pr_url="https://github.com/${REPO}/pull/${PR_NUMBER}"
-          commit_url="https://github.com/${REPO}/commit/${HEAD_SHA}"
-
-          gh api --paginate "repos/${REPO}/pulls/${PR_NUMBER}/comments?per_page=100" --jq '.[] | @json' > "${comments}"
-
-          python3 - "${before_ids}" "${comments}" "${review_data}" "${HEAD_SHA}" "${pr_url}" "${commit_url}" "${short_sha}" <<'PY'
-          import json
-          import sys
-          from pathlib import Path
-
-          before_ids = set(json.loads(sys.argv[1] or "[]"))
-          comments_path = Path(sys.argv[2])
-          output_path = Path(sys.argv[3])
-          head_sha = sys.argv[4]
-          pr_url = sys.argv[5]
-          commit_url = sys.argv[6]
-          short_sha = sys.argv[7]
-
-          comments = []
-          for line in comments_path.read_text().splitlines():
-              if line.strip():
-                  comments.append(json.loads(line))
-
-          new_comments = [
-              item for item in comments
-              if item.get("user", {}).get("login") == "github-actions[bot]"
-              and item.get("id") not in before_ids
-              and item.get("commit_id") == head_sha
-          ]
-          suggestions = []
-          for item in new_comments:
-              body = (item.get("body") or "").strip()
-              first_line = next((line.strip() for line in body.splitlines() if line.strip()), "")
-              suggestions.append({
-                  "path": item.get("path"),
-                  "line": item.get("line") or item.get("start_line"),
-                  "url": item.get("html_url"),
-                  "summary": first_line[:500],
-                  "body": body[:1500],
-              })
-
-          output_path.write_text(json.dumps({
-              "pr_url": pr_url,
-              "commit_url": commit_url,
-              "short_sha": short_sha,
-              "suggestions": suggestions,
-          }, ensure_ascii=False))
-          PY
-
-          python3 - "${review_data}" "${payload}" <<'PY'
-          import json
-          import os
-          import textwrap
-          import urllib.error
-          import urllib.request
-          from pathlib import Path
-
-          review_data = json.loads(Path(os.sys.argv[1]).read_text())
-          payload_path = Path(os.sys.argv[2])
-          suggestions = review_data["suggestions"]
-          marker = "<!-- pr-agent-code-review-summary -->"
-          summary_language = os.environ.get("SUMMARY_LANGUAGE") or "中文"
-          use_chinese = summary_language != "English"
-
-          def fallback_summary() -> str:
-              if not suggestions:
-                  if use_chinese:
-                      return "已审查本次变更，未发现需要进一步反馈或调整的问题。"
-                  return "Reviewed this change and found no further feedback or required adjustments."
-              if use_chinese:
-                  lines = [f"本轮代码审查新增 {len(suggestions)} 条行内建议，建议优先查看以下位置："]
-              else:
-                  noun = "suggestion" if len(suggestions) == 1 else "suggestions"
-                  lines = [f"This review added {len(suggestions)} inline {noun}. Consider reviewing these locations first:"]
-              for item in suggestions[:5]:
-                  location = f"{item.get('path')}:{item.get('line')}" if item.get("line") else str(item.get("path"))
-                  summary = item.get("summary") or "查看行内建议"
-                  url = item.get("url")
-                  if use_chinese:
-                      lines.append(f"- [{location}]({url})：{summary}" if url else f"- {location}：{summary}")
-                  else:
-                      lines.append(f"- [{location}]({url}): {summary}" if url else f"- {location}: {summary}")
-              if len(suggestions) > 5:
-                  if use_chinese:
-                      lines.append(f"- 其余 {len(suggestions) - 5} 条请在 Files changed 的行内评论中查看。")
-                  else:
-                      lines.append(f"- Review the remaining {len(suggestions) - 5} inline comments in Files changed.")
-              return "\n".join(lines)
-
-          def llm_summary() -> str | None:
-              if not suggestions:
-                  return None
-
-              api_base = (os.environ.get("OPENAI_API_BASE") or "").rstrip("/")
-              api_key = os.environ.get("OPENAI_KEY") or ""
-              model = os.environ.get("SUMMARY_MODEL") or "gpt-5.5"
-              if not api_base or not api_key:
-                  return None
-
-              if use_chinese:
-                  task = (
-                      "你是独立的代码审查摘要助手。只基于本轮已经发布的行内审查意见做简短汇总，"
-                      "不新增审查结论，不替维护者判断 PR 是否可以合并。请用中文 Markdown 输出："
-                      "第一段说明本轮审查已完成，并已在行内留下需要关注的建议；如有建议，"
-                      "概括 1-3 个主要关注点，使用“建议关注”“可能影响”“可优先查看”等中立表述。"
-                      "不要输出表格，不要复述所有文件，不要使用“可以合并”“不建议合并”“阻塞合并”"
-                      "“先处理后再合入”等合并裁决措辞。"
-                  )
-                  system_prompt = "你是严谨、中立的代码审查摘要助手。输出中文 Markdown，简洁自然。"
-              else:
-                  task = (
-                      "You are an independent code review summarizer. Summarize only the inline review comments "
-                      "already posted in this run; do not add new review conclusions or decide whether the PR should merge. "
-                      "Write concise Markdown in English. Start by noting that review completed and inline suggestions "
-                      "were left; then summarize 1-3 main points using neutral phrasing such as \"consider\", "
-                      "\"may affect\", or \"worth reviewing\". Do not output tables, list every file, or use merge-gate "
-                      "wording such as \"ready to merge\", \"do not merge\", \"blocks merging\", or \"must be fixed before merge\"."
-                  )
-                  system_prompt = "You are a rigorous, neutral code review summarizer. Write concise English Markdown."
-
-              user_content = {
-                  "task": task,
-                  "suggestions": suggestions[:10],
-              }
-
-              request_body = {
-                  "model": model,
-                  "messages": [
-                      {"role": "system", "content": system_prompt},
-                      {"role": "user", "content": json.dumps(user_content, ensure_ascii=False)},
-                  ],
-                  "temperature": 0.2,
-              }
-              request = urllib.request.Request(
-                  f"{api_base}/chat/completions",
-                  data=json.dumps(request_body).encode("utf-8"),
-                  headers={
-                      "Authorization": f"Bearer {api_key}",
-                      "Content-Type": "application/json",
-                  },
-                  method="POST",
-              )
-              try:
-                  with urllib.request.urlopen(request, timeout=60) as response:
-                      data = json.loads(response.read().decode("utf-8"))
-                  content = data["choices"][0]["message"]["content"].strip()
-                  return content or None
-              except (KeyError, TimeoutError, urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError):
-                  return None
-
-          summary = llm_summary() or fallback_summary()
-          commit_label = "审查提交：" if use_chinese else "Reviewed commit:"
-          body = textwrap.dedent(f"""\
-          {marker}
-          ## Code Review
-
-          {summary}
-
-          {commit_label} [{review_data["short_sha"]}]({review_data["commit_url"]})
-          """)
-          payload_path.write_text(json.dumps({"body": body}, ensure_ascii=False))
-          PY
-
-          new_comment_id="$(gh api --method POST "repos/${REPO}/issues/${PR_NUMBER}/comments" --input "${payload}" --jq '.id')"
-          comment_ids="$(gh api --paginate "repos/${REPO}/issues/${PR_NUMBER}/comments?per_page=100" --jq ".[] | select(.id != ${new_comment_id} and .user.login == \"github-actions[bot]\" and ((.body | startswith(\"<!-- pr-agent-code-review-summary -->\")) or (.body | startswith(\"<!-- pr-agent-update-notification -->\")))) | .id")"
-
-          if [ -z "${comment_ids}" ]; then
-            echo "No previous PR-Agent code review summary to clean."
-            exit 0
-          fi
-
-          while IFS= read -r comment_id; do
-            [ -z "${comment_id}" ] && continue
-            gh api --method DELETE "repos/${REPO}/issues/comments/${comment_id}" >/dev/null
-          done <<< "${comment_ids}"
-
-      - name: Restore PR-Agent description markers on failure
-        if: >-
-          failure() &&
-          steps.prepare_description.outputs.body_changed == 'true' &&
-          steps.pr_language.outputs.skip_pr_agent != 'true'
-        env:
-          GH_TOKEN: ${{ github.token }}
-          REPO: ${{ github.repository }}
-          PR_NUMBER: ${{ github.event.pull_request.number || github.event.issue.number }}
-        run: |
-          set -euo pipefail
-          body_backup="${RUNNER_TEMP}/pr-agent-body-before-describe.md"
-          placeholder_body="${RUNNER_TEMP}/pr-agent-body-with-placeholder.md"
-          current_body="$(mktemp)"
-          payload="$(mktemp)"
-
-          if [ ! -s "${body_backup}" ] || [ ! -s "${placeholder_body}" ]; then
-            echo "No PR body backup found."
-            exit 0
-          fi
-
-          gh api "repos/${REPO}/pulls/${PR_NUMBER}" --jq '.body // ""' > "${current_body}"
-          python3 - "${body_backup}" "${placeholder_body}" "${current_body}" "${payload}" <<'PY'
-          import json
-          import re
-          import sys
-          from pathlib import Path
-
-          backup_body = Path(sys.argv[1]).read_text()
-          placeholder_body = Path(sys.argv[2]).read_text()
-          current_body = Path(sys.argv[3]).read_text()
-          payload_path = Path(sys.argv[4])
-
-          start = "<!-- pr-agent-summary:start -->"
-          end = "<!-- pr-agent-summary:end -->"
-          heading_re = re.compile(r"(?im)^##\s+(PR-Agent\s+摘要|PR-Agent Summary)\s*\n\s*")
-
-          def find_block(body: str) -> tuple[int, int] | None:
               start_index = body.find(start)
-              end_index = body.find(end)
-              if start_index < 0 or end_index <= start_index:
-                  return None
-              end_index += len(end)
-              heading_start = start_index
-              prefix = body[:start_index]
-              matches = list(heading_re.finditer(prefix))
-              if matches:
-                  last = matches[-1]
-                  if prefix[last.end():].strip() == "":
-                      heading_start = last.start()
-              return heading_start, end_index
-
-          current_block = find_block(current_body)
-          placeholder_block = find_block(placeholder_body)
-          backup_block = find_block(backup_body)
-          if not current_block or not placeholder_block:
-              print("No PR-Agent summary block to restore.")
-              raise SystemExit(0)
-
-          current_section = current_body[current_block[0]:current_block[1]]
-          placeholder_section = placeholder_body[placeholder_block[0]:placeholder_block[1]]
-          if "pr_agent:summary" not in current_section:
-              print("No visible PR-Agent summary placeholder to restore.")
-              raise SystemExit(0)
-          if current_section != placeholder_section:
-              print("Current PR-Agent summary block changed; skip restore.")
-              raise SystemExit(0)
-
-          restored_section = backup_body[backup_block[0]:backup_block[1]] if backup_block else ""
-          next_body = current_body[:current_block[0]] + restored_section + current_body[current_block[1]:]
-          next_body = re.sub(r"\n{4,}", "\n\n\n", next_body).rstrip() + "\n"
-          payload_path.write_text(json.dumps({"body": next_body}, ensure_ascii=False))
+              end_index = body.find(end, start_index + len(start))
+              updated = body[:start_index] + f"{start}\npr_agent:summary\n{end}" + body[end_index + len(end):]
+          elif start_index >= 0 or end in body:
+              updated = body
+          elif body.strip():
+              updated = f"{body.rstrip()}\n\n{block}\n"
+          else:
+              updated = f"{block}\n"
+          if updated != body:
+              with open(sys.argv[1], "w", encoding="utf-8") as handle:
+                  json.dump({"body": updated}, handle, ensure_ascii=False)
           PY
           if [ -s "${payload}" ]; then
             gh api --method PATCH "repos/${REPO}/pulls/${PR_NUMBER}" --input "${payload}" >/dev/null
           fi
+
+      - name: Update PR description
+        if: >-
+          steps.pr_context.outputs.skip_pr_agent != 'true' &&
+          steps.pr_context.outputs.changed_files != '0' &&
+          (
+            github.event_name == 'pull_request_target' ||
+            github.event.comment.body == '/describe' ||
+            startsWith(github.event.comment.body, '/describe ')
+          )
+        uses: docker://pragent/pr-agent:0.39.0-github_action@sha256:b253845caa8c7ff5ce8be78f32996647982bdd4890826a962b78eff2e385a825
+        env:
+          GITHUB_TOKEN: ${{ github.token }}
+          OPENAI_KEY: ${{ secrets.OPENAI_KEY }}
+          OPENAI.API_BASE: ${{ secrets.OPENAI_API_BASE }}
+          config.model: 'gpt-5.6-terra'
+          config.fallback_models: '["gpt-5.5", "gpt-5.4"]'
+          config.custom_model_max_tokens: '1050000'
+          config.reasoning_effort: 'medium'
+          config.ai_timeout: '900'
+          config.response_language: ${{ steps.pr_context.outputs.response_language }}
+          config.large_patch_policy: 'clip'
+          config.ignore_pr_title: '["^\\[Auto\\]", "^Auto"]'
+          config.ignore_pr_labels: '["skip pr-agent"]'
+          github_action_config.auto_review: 'false'
+          github_action_config.auto_describe: 'true'
+          github_action_config.auto_improve: 'false'
+          github_action_config.pr_actions: '["opened", "reopened", "ready_for_review", "review_requested"]'
+          github_action_config.handle_push_trigger: 'true'
+          github_action_config.push_commands: '["/describe"]'
+          pr_description.generate_ai_title: 'false'
+          pr_description.publish_labels: 'false'
+          pr_description.publish_description_as_comment: 'false'
+          pr_description.publish_description_as_comment_persistent: 'false'
+          pr_description.enable_pr_diagram: 'false'
+          pr_description.enable_pr_type: 'false'
+          pr_description.enable_help_text: 'false'
+          pr_description.enable_help_comment: 'false'
+          pr_description.enable_semantic_files_types: 'false'
+          pr_description.collapsible_file_list: 'adaptive'
+          pr_description.add_original_user_description: 'true'
+          pr_description.use_description_markers: 'true'
+          pr_description.final_update_message: 'false'
+          pr_description.extra_instructions: |
+            Match the configured response language.
+            Summarize the change goal, key implementation details, compatibility impact, tests, and notable risks.
+            Use 2-4 bullets for small pull requests and 4-8 bullets for larger changes.
+            Avoid file lists and local command transcripts.
+
+      - name: Remove unfilled PR-Agent description marker
+        if: >-
+          always() &&
+          steps.pr_context.outputs.skip_pr_agent != 'true' &&
+          (
+            github.event_name == 'pull_request_target' ||
+            github.event.comment.body == '/describe' ||
+            startsWith(github.event.comment.body, '/describe ')
+          )
+        env:
+          GH_TOKEN: ${{ github.token }}
+          REPO: ${{ github.repository }}
+          PR_NUMBER: ${{ github.event.pull_request.number || github.event.issue.number }}
+        run: |
+          set -euo pipefail
+          payload="$(mktemp)"
+          python3 - "${payload}" <<'PY'
+          import json
+          import os
+          import re
+          import subprocess
+          import sys
+
+          pull = json.loads(subprocess.check_output(
+              ["gh", "api", f"repos/{os.environ['REPO']}/pulls/{os.environ['PR_NUMBER']}"],
+              text=True,
+          ))
+          body = pull.get("body") or ""
+          placeholder = "<!-- pr-agent-summary:start -->\npr_agent:summary\n<!-- pr-agent-summary:end -->"
+          owned_block = re.compile(
+              r"(?ims)^##\s+(?:PR-Agent\s+摘要|PR-Agent\s+Summary)\s*\n\s*"
+              r"<!-- pr-agent-summary:start -->\s*pr_agent:summary\s*<!-- pr-agent-summary:end -->\s*"
+          )
+          if placeholder in body:
+              updated = owned_block.sub("", body).rstrip()
+              if updated != body:
+                  with open(sys.argv[1], "w", encoding="utf-8") as handle:
+                      json.dump({"body": updated}, handle, ensure_ascii=False)
+          PY
+          if [ -s "${payload}" ]; then
+            gh api --method PATCH "repos/${REPO}/pulls/${PR_NUMBER}" --input "${payload}" >/dev/null
+          fi
+
+      - name: Analyze PR review
+        id: review_analysis
+        if: >-
+          steps.pr_context.outputs.skip_pr_agent != 'true' &&
+          (
+            github.event_name == 'pull_request_target' ||
+            github.event.comment.body == '/review' ||
+            startsWith(github.event.comment.body, '/review ')
+          )
+        uses: docker://pragent/pr-agent:0.39.0-github_action@sha256:b253845caa8c7ff5ce8be78f32996647982bdd4890826a962b78eff2e385a825
+        env:
+          GITHUB_TOKEN: ${{ github.token }}
+          OPENAI_KEY: ${{ secrets.OPENAI_KEY }}
+          OPENAI.API_BASE: ${{ secrets.OPENAI_API_BASE }}
+          config.model: ${{ github.event_name == 'issue_comment' && 'gpt-5.6-sol' || 'gpt-5.6-terra' }}
+          config.fallback_models: '["gpt-5.5", "gpt-5.4"]'
+          config.custom_model_max_tokens: '1050000'
+          config.reasoning_effort: 'xhigh'
+          config.ai_timeout: '900'
+          config.response_language: ${{ steps.pr_context.outputs.response_language }}
+          config.large_patch_policy: 'clip'
+          config.ignore_pr_title: '["^\\[Auto\\]", "^Auto"]'
+          config.ignore_pr_labels: '["skip pr-agent"]'
+          config.publish_output: 'false'
+          github_action_config.auto_review: 'true'
+          github_action_config.auto_describe: 'false'
+          github_action_config.auto_improve: 'false'
+          github_action_config.pr_actions: '["opened", "reopened", "ready_for_review", "review_requested"]'
+          github_action_config.handle_push_trigger: 'true'
+          github_action_config.push_commands: '["/review"]'
+          github_action_config.enable_output: 'true'
+          pr_reviewer.num_max_findings: '4'
+          pr_reviewer.require_score_review: 'false'
+          pr_reviewer.require_tests_review: 'false'
+          pr_reviewer.require_security_review: 'false'
+          pr_reviewer.require_estimate_effort_to_review: 'false'
+          pr_reviewer.require_estimate_contribution_time_cost: 'false'
+          pr_reviewer.require_can_be_split_review: 'false'
+          pr_reviewer.require_todo_scan: 'false'
+          pr_reviewer.require_ticket_analysis_review: 'false'
+          pr_reviewer.enable_review_labels_effort: 'false'
+          pr_reviewer.enable_review_labels_security: 'false'
+          pr_reviewer.extra_instructions: |
+            Return key_issues_to_review only for concrete behavior defects introduced by this pull request.
+            Each finding must identify the affected behavior, a reachable trigger, and the existing contract or invariant it violates.
+            Use issue_content to state the smallest correction boundary, not a code patch.
+            Do not report style preferences, comments, refactors, architecture alternatives, speculative races, extra hardening, optional tests, or hypothetical concerns.
+            Return no findings when the evidence is incomplete.
+
+      - name: Publish review comments and summary
+        if: >-
+          steps.pr_context.outputs.skip_pr_agent != 'true' &&
+          steps.review_analysis.outcome == 'success' &&
+          (
+            github.event_name == 'pull_request_target' ||
+            github.event.comment.body == '/review' ||
+            startsWith(github.event.comment.body, '/review ')
+          )
+        env:
+          GH_TOKEN: ${{ github.token }}
+          REPO: ${{ github.repository }}
+          PR_NUMBER: ${{ github.event.pull_request.number || github.event.issue.number }}
+          REVIEWED_HEAD_SHA: ${{ steps.pr_context.outputs.head_sha }}
+          CHANGED_FILES: ${{ steps.pr_context.outputs.changed_files }}
+          RESPONSE_LANGUAGE: ${{ steps.pr_context.outputs.response_language }}
+          REVIEW_JSON: ${{ steps.review_analysis.outputs.review }}
+        run: |
+          set -euo pipefail
+          current_head_sha="$(gh api "repos/${REPO}/pulls/${PR_NUMBER}" --jq '.head.sha')"
+          if [ "${current_head_sha}" != "${REVIEWED_HEAD_SHA}" ]; then
+            echo "PR head changed during analysis; skip stale review publication."
+            exit 0
+          fi
+
+          review_payload="$(mktemp)"
+          python3 - "${review_payload}" <<'PY'
+          import hashlib
+          import json
+          import os
+          import re
+          import subprocess
+          import sys
+          from urllib.parse import quote
+
+          review_raw = os.environ.get("REVIEW_JSON") or "{}"
+          review = json.loads(review_raw)
+          if not review_raw.strip() or review == {}:
+              if int(os.environ.get("CHANGED_FILES") or 0):
+                  raise SystemExit("Review analysis produced no structured output for a non-empty PR.")
+              review = {}
+
+          repo = os.environ["REPO"]
+          number = os.environ["PR_NUMBER"]
+          head_sha = os.environ["REVIEWED_HEAD_SHA"]
+          language = os.environ.get("RESPONSE_LANGUAGE") or "zh-CN"
+
+          def paged(endpoint):
+              result = json.loads(subprocess.check_output(
+                  ["gh", "api", "--paginate", "--slurp", endpoint], text=True
+              ))
+              if result and all(isinstance(page, list) for page in result):
+                  return [item for page in result for item in page]
+              return result
+
+          files = paged(f"repos/{repo}/pulls/{number}/files?per_page=100")
+          comments = paged(f"repos/{repo}/pulls/{number}/comments?per_page=100")
+          reviews = paged(f"repos/{repo}/pulls/{number}/reviews?per_page=100")
+
+          hunk_pattern = re.compile(r"^@@ -\d+(?:,\d+)? \+(\d+)(?:,\d+)? @@")
+          changed_lines = {}
+          for file_data in files:
+              path = str(file_data.get("filename") or "")
+              line = None
+              lines = set()
+              for patch_line in (file_data.get("patch") or "").splitlines():
+                  hunk = hunk_pattern.match(patch_line)
+                  if hunk:
+                      line = int(hunk.group(1))
+                      continue
+                  if line is None or patch_line.startswith("\\"):
+                      continue
+                  if patch_line.startswith("+") and not patch_line.startswith("+++"):
+                      lines.add(line)
+                      line += 1
+                  elif patch_line.startswith("-") and not patch_line.startswith("---"):
+                      continue
+                  else:
+                      line += 1
+              changed_lines[path] = lines
+
+          def fingerprint(path, line):
+              normalized = "\n".join((path, str(line)))
+              return hashlib.sha256(normalized.encode("utf-8")).hexdigest()[:16]
+
+          current_fingerprints = set()
+          current_locations = set()
+          marker_pattern = re.compile(r"<!-- pr-agent-review:([0-9a-f]{16}) -->")
+          for comment in comments:
+              if comment.get("user", {}).get("login") != "github-actions[bot]":
+                  continue
+              if comment.get("line") is None:
+                  continue
+              match = marker_pattern.search(str(comment.get("body") or ""))
+              if match:
+                  current_fingerprints.add(match.group(1))
+                  path = str(comment.get("path") or "")
+                  try:
+                      line = int(comment.get("line") or 0)
+                  except (TypeError, ValueError):
+                      line = 0
+                  if path and line > 0:
+                      current_locations.add((path, line))
+
+          def code_url(path, line):
+              return f"https://github.com/{repo}/blob/{head_sha}/{quote(path, safe='/')}#L{line}"
+
+          issues = review.get("key_issues_to_review") or []
+          findings = []
+          seen = set()
+          for issue in issues:
+              if not isinstance(issue, dict):
+                  continue
+              path = str(issue.get("relevant_file") or "").strip()
+              header = str(issue.get("issue_header") or "").strip()
+              content = str(issue.get("issue_content") or "").strip()
+              try:
+                  line = int(issue.get("start_line") or 0)
+              except (TypeError, ValueError):
+                  line = 0
+              if not path or not header or not content or line < 1:
+                  continue
+              finding_key = (path, line, header.lower(), " ".join(content.split()).lower())
+              if finding_key in seen:
+                  continue
+              seen.add(finding_key)
+              findings.append({
+                  "path": path,
+                  "line": line,
+                  "header": header,
+                  "content": content,
+                  "fingerprint": fingerprint(path, line),
+              })
+
+          new_comments = []
+          for finding in findings:
+              if finding["line"] not in changed_lines.get(finding["path"], set()):
+                  continue
+              if finding["fingerprint"] in current_fingerprints or (finding["path"], finding["line"]) in current_locations:
+                  continue
+              new_comments.append({
+                  "path": finding["path"],
+                  "line": finding["line"],
+                  "side": "RIGHT",
+                  "body": "\n".join([
+                      f"<!-- pr-agent-review:{finding['fingerprint']} -->",
+                      f"**{finding['header']}**",
+                      "",
+                      finding["content"],
+                  ]),
+              })
+
+          marker = "<!-- pr-agent-review-summary -->"
+          short_sha = head_sha[:7]
+          commit_url = f"https://github.com/{repo}/commit/{head_sha}"
+          chinese = language == "zh-CN"
+          lines = [marker, "## PR-Agent Code Review", ""]
+          if findings:
+              for finding in findings:
+                  location = f"{finding['path']}:{finding['line']}"
+                  concise = " ".join(finding["content"].split())[:360]
+                  separator = "：" if chinese else ":"
+                  lines.append(f"- [{location}]({code_url(finding['path'], finding['line'])}){separator} **{finding['header']}** - {concise}")
+          elif chinese:
+              lines.append("本次变更无需提出审查意见，暂无其他反馈。")
+          else:
+              lines.append("There are no review comments for the current changes. I have no additional feedback to provide.")
+          lines.extend([
+              "",
+              f"审查提交：[{short_sha}]({commit_url})" if chinese else f"Reviewed commit: [{short_sha}]({commit_url})",
+              "",
+          ])
+          payload = {
+              "body": "\n".join(lines),
+              "commit_id": head_sha,
+              "event": "COMMENT",
+          }
+          if new_comments:
+              payload["comments"] = new_comments
+          has_matching_summary = not new_comments and any(
+              existing.get("user", {}).get("login") == "github-actions[bot]"
+              and existing.get("commit_id") == head_sha
+              and str(existing.get("body") or "") == payload["body"]
+              for existing in reviews
+          )
+          # 同一提交的手工重审仍会完成分析；完全相同的结果不重复发布 Review。
+          if not has_matching_summary:
+              with open(sys.argv[1], "w", encoding="utf-8") as handle:
+                  json.dump(payload, handle, ensure_ascii=False)
+          PY
+
+          latest_head_sha="$(gh api "repos/${REPO}/pulls/${PR_NUMBER}" --jq '.head.sha')"
+          if [ "${latest_head_sha}" != "${REVIEWED_HEAD_SHA}" ]; then
+            echo "PR head changed while rendering review; skip stale review publication."
+            exit 0
+          fi
+
+          if [ -s "${review_payload}" ]; then
+            gh api --method POST "repos/${REPO}/pulls/${PR_NUMBER}/reviews" --input "${review_payload}" >/dev/null
+          fi
+          old_summary_ids="$(gh api --paginate "repos/${REPO}/issues/${PR_NUMBER}/comments?per_page=100" --jq ".[] | select(.user.login == \"github-actions[bot]\" and ((.body | startswith(\"<!-- pr-agent-update-notification -->\")) or (.body | startswith(\"<!-- pr-agent-code-review-summary -->\")) or (.body | startswith(\"<!-- pr-agent-review-summary -->\")) or (.body | startswith(\"<!-- pr-agent-lab:review -->\")))) | .id")"
+          while IFS= read -r comment_id; do
+            [ -z "${comment_id}" ] && continue
+            gh api --method DELETE "repos/${REPO}/issues/comments/${comment_id}" >/dev/null
+          done <<< "${old_summary_ids}"
+
+      - name: Answer PR question
+        if: >-
+          steps.pr_context.outputs.skip_pr_agent != 'true' &&
+          github.event_name == 'issue_comment' &&
+          (
+            github.event.comment.body == '/ask' ||
+            startsWith(github.event.comment.body, '/ask ')
+          )
+        uses: docker://pragent/pr-agent:0.39.0-github_action@sha256:b253845caa8c7ff5ce8be78f32996647982bdd4890826a962b78eff2e385a825
+        env:
+          GITHUB_TOKEN: ${{ github.token }}
+          OPENAI_KEY: ${{ secrets.OPENAI_KEY }}
+          OPENAI.API_BASE: ${{ secrets.OPENAI_API_BASE }}
+          config.model: 'gpt-5.6-terra'
+          config.fallback_models: '["gpt-5.5", "gpt-5.4"]'
+          config.custom_model_max_tokens: '1050000'
+          config.reasoning_effort: 'high'
+          config.ai_timeout: '900'
+          config.response_language: ${{ steps.pr_context.outputs.response_language }}
+          config.large_patch_policy: 'clip'
+          config.ignore_pr_title: '["^\\[Auto\\]", "^Auto"]'
+          config.ignore_pr_labels: '["skip pr-agent"]'
+          github_action_config.auto_review: 'false'
+          github_action_config.auto_describe: 'false'
+          github_action_config.auto_improve: 'false'
diff --git a/app/db/__init__.py b/app/db/__init__.py
index efc09916..43843511 100644
--- a/app/db/__init__.py
+++ b/app/db/__init__.py
@@ -1,12 +1,60 @@
 import asyncio
 from typing import Any, Generator, List, Optional, Self, Tuple, AsyncGenerator, Union
 
-from sqlalchemy import NullPool, QueuePool, and_, create_engine, inspect, text, select, delete, Column, Integer, \
+from sqlalchemy import NullPool, QueuePool, and_, create_engine, event, inspect, text, select, delete, Column, Integer, \
     Sequence, Identity
+from sqlalchemy.engine import Engine as SQLAlchemyEngine, ExceptionContext
 from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
 from sqlalchemy.orm import Session, as_declarative, declared_attr, scoped_session, sessionmaker
 
 from app.core.config import settings
+from app.log import logger
+
+
+def _database_error_metadata(error: BaseException) -> Optional[dict[str, Any]]:
+    """提取 SQLite 与 PostgreSQL 驱动提供的稳定错误分类字段。"""
+    metadata = {"error_type": type(error).__name__}
+
+    # DBAPI 驱动字段并不共享统一类型，动态读取可同时兼容 sqlite3、psycopg2 与 asyncpg。
+    sqlite_errorcode = getattr(error, "sqlite_errorcode", None)
+    sqlite_errorname = getattr(error, "sqlite_errorname", None)
+    if sqlite_errorcode is not None or sqlite_errorname:
+        if sqlite_errorcode is not None:
+            metadata["error_code"] = sqlite_errorcode
+        if sqlite_errorname:
+            metadata["error_name"] = sqlite_errorname
+        return metadata
+
+    sqlstate = getattr(error, "sqlstate", None) or getattr(error, "pgcode", None)
+    if not sqlstate:
+        sqlstate = getattr(getattr(error, "diag", None), "sqlstate", None)
+    if sqlstate:
+        metadata["sqlstate"] = sqlstate
+        return metadata
+
+    return None
+
+
+def _log_database_error(exception_context: ExceptionContext) -> None:
+    """记录非敏感驱动错误码，并保持 SQLAlchemy 原有异常传播。"""
+    metadata = _database_error_metadata(exception_context.original_exception)
+    if not metadata:
+        return
+
+    dialect = exception_context.dialect
+    fields = {
+        "database": dialect.name,
+        "driver": dialect.driver,
+        **metadata,
+    }
+    logger.error(
+        "数据库驱动异常：" + ", ".join(f"{key}={value}" for key, value in fields.items())
+    )
+
+
+def _register_database_error_logging(engine: SQLAlchemyEngine) -> None:
+    """为主程序 Engine 注册统一的底层驱动错误诊断。"""
+    event.listen(engine, "handle_error", _log_database_error)
 
 
 def get_id_column():
@@ -71,6 +119,7 @@ def _get_sqlite_engine(is_async: bool = False):
 
         # 创建数据库引擎
         engine = create_engine(**_db_kwargs)
+        _register_database_error_logging(engine)
 
         # 设置WAL模式
         _journal_mode = "WAL" if settings.DB_WAL_ENABLE else "DELETE"
@@ -91,6 +140,7 @@ def _get_sqlite_engine(is_async: bool = False):
         }
         # 创建异步数据库引擎
         async_engine = create_async_engine(**_db_kwargs)
+        _register_database_error_logging(async_engine.sync_engine)
 
         # 设置WAL模式
         _journal_mode = "WAL" if settings.DB_WAL_ENABLE else "DELETE"
@@ -146,6 +196,7 @@ def _get_postgresql_engine(is_async: bool = False):
 
         # 创建数据库引擎
         engine = create_engine(**_db_kwargs)
+        _register_database_error_logging(engine)
         print(f"PostgreSQL database connected to {settings.DB_POSTGRESQL_TARGET}/{settings.DB_POSTGRESQL_DATABASE}")
 
         return engine
@@ -163,6 +214,7 @@ def _get_postgresql_engine(is_async: bool = False):
         }
         # 创建异步数据库引擎
         async_engine = create_async_engine(**_db_kwargs)
+        _register_database_error_logging(async_engine.sync_engine)
         print(f"Async PostgreSQL database connected to {settings.DB_POSTGRESQL_TARGET}/{settings.DB_POSTGRESQL_DATABASE}")
 
         return async_engine
diff --git a/app/helper/doh.py b/app/helper/doh.py
index 620be393..be1e69b4 100644
--- a/app/helper/doh.py
+++ b/app/helper/doh.py
@@ -18,8 +18,10 @@ from app.log import logger
 from app.utils.mixins import ConfigReloadMixin
 from app.utils.singleton import Singleton
 
-# 定义一个全局线程池执行器
-_executor = concurrent.futures.ThreadPoolExecutor()
+# DoH 关闭时需要释放线程池；保持惰性创建可避免未启用 DoH 时占用进程级资源
+_executor: Optional[concurrent.futures.ThreadPoolExecutor] = None
+_executor_lock = Lock()
+_doh_enabled = False
 
 # 定义默认的DoH配置
 _doh_timeout = 5
@@ -29,11 +31,21 @@ _doh_lock = Lock()
 _orig_getaddrinfo = socket.getaddrinfo
 
 
+def _get_executor_locked() -> concurrent.futures.ThreadPoolExecutor:
+    """在持有执行器锁时按需获取 DoH 查询线程池"""
+    global _executor
+    if _executor is None:
+        _executor = concurrent.futures.ThreadPoolExecutor()
+    return _executor
+
+
 def enable_doh(enable: bool) -> None:
     """
     对 socket.getaddrinfo 进行补丁
     """
 
+    global _doh_enabled
+
     def _patched_getaddrinfo(host: str, *args, **kwargs):
         """
         socket.getaddrinfo的补丁版本。
@@ -47,9 +59,15 @@ def enable_doh(enable: bool) -> None:
             logger.info(f"已解析 [{host}] 为 [{ip}] (缓存)")
             return _orig_getaddrinfo(ip, *args, **kwargs)
         # 使用DoH解析主机
-        futures = []
-        for resolver in settings.DOH_RESOLVERS.split(","):
-            futures.append(_executor.submit(_doh_query, resolver, host))
+        with _executor_lock:
+            if not _doh_enabled:
+                return _orig_getaddrinfo(host, *args, **kwargs)
+            executor = _get_executor_locked()
+            # 一次解析的任务必须在同一临界区提交完，避免关闭过程中部分任务落入新线程池
+            futures = [
+                executor.submit(_doh_query, resolver, host)
+                for resolver in settings.DOH_RESOLVERS.split(",")
+            ]
         for future in concurrent.futures.as_completed(futures):
             ip = future.result()
             if ip is not None:
@@ -60,11 +78,9 @@ def enable_doh(enable: bool) -> None:
                 break
         return _orig_getaddrinfo(host, *args, **kwargs)
 
-    if enable:
-        # 替换 socket.getaddrinfo 方法
-        socket.getaddrinfo = _patched_getaddrinfo
-    else:
-        socket.getaddrinfo = _orig_getaddrinfo
+    with _executor_lock:
+        _doh_enabled = enable
+        socket.getaddrinfo = _patched_getaddrinfo if enable else _orig_getaddrinfo
 
 
 class DohHelper(ConfigReloadMixin, metaclass=Singleton):
@@ -77,14 +93,31 @@ class DohHelper(ConfigReloadMixin, metaclass=Singleton):
         enable_doh(settings.DOH_ENABLE)
 
     def on_config_changed(self) -> None:
+        if not settings.DOH_ENABLE:
+            self.shutdown()
+            return
         with _doh_lock:
             # DOH配置有变动的情况下，清空缓存
             _doh_cache.clear()
-        enable_doh(settings.DOH_ENABLE)
+        enable_doh(True)
 
     def get_reload_name(self) -> str:
         return 'DoH'
 
+    def shutdown(self) -> None:
+        """恢复系统 DNS 并释放 DoH 查询线程池"""
+        global _executor, _doh_enabled
+        with _executor_lock:
+            _doh_enabled = False
+            socket.getaddrinfo = _orig_getaddrinfo
+            executor = _executor
+            _executor = None
+        with _doh_lock:
+            _doh_cache.clear()
+        if executor:
+            executor.shutdown(wait=True)
+
+
 def _doh_query(resolver: str, host: str) -> Optional[str]:
     """
     使用给定的DoH解析器查询给定主机的IP地址。
diff --git a/app/helper/message.py b/app/helper/message.py
index 3fb9be12..e185dde9 100644
--- a/app/helper/message.py
+++ b/app/helper/message.py
@@ -605,6 +605,7 @@ class MessageQueueManager(metaclass=SingletonClass):
         self.check_interval = check_interval
 
         self._running = True
+        self._stop_event = threading.Event()
         self.thread = threading.Thread(target=self._monitor_loop, daemon=True)
         self.thread.start()
 
@@ -752,13 +753,15 @@ class MessageQueueManager(metaclass=SingletonClass):
                         logger.info(f"队列剩余消息：{self.queue.qsize()}")
                     except queue.Empty:
                         break
-            time.sleep(self.check_interval)
+            if self._stop_event.wait(self.check_interval):
+                break
 
     def stop(self) -> None:
         """
         停止队列管理器
         """
         self._running = False
+        self._stop_event.set()
         logger.info("正在停止消息队列...")
         self.thread.join()
         logger.info("消息队列已停止")
@@ -841,7 +844,8 @@ def stop_message():
     """
     停止消息服务
     """
-    # 停止消息队列
-    MessageQueueManager().stop()
-    # 关闭消息演染器
-    TemplateHelper().close()
+    # 只关闭已启动的服务，避免清理路径反向创建后台线程和缓存
+    if queue_manager := MessageQueueManager.get_existing_instance():
+        queue_manager.stop()
+    if template_helper := TemplateHelper.get_existing_instance():
+        template_helper.close()
diff --git a/app/log.py b/app/log.py
index 7a922dc8..85957b57 100644
--- a/app/log.py
+++ b/app/log.py
@@ -124,7 +124,7 @@ class NonBlockingFileHandler:
     """
     _instance = None
     _lock = threading.Lock()
-    _rotating_handlers = {}
+    _stop_sentinel = object()
 
     def __new__(cls):
         if cls._instance is None:
@@ -138,6 +138,9 @@ class NonBlockingFileHandler:
             return
 
         self._initialized = True
+        self._state_lock = threading.RLock()
+        self._handlers_lock = threading.Lock()
+        self._rotating_handlers = {}
         self._write_queue = queue.Queue(maxsize=log_settings.ASYNC_FILE_QUEUE_SIZE)
         self._executor = ThreadPoolExecutor(max_workers=log_settings.ASYNC_FILE_WORKERS,
                                             thread_name_prefix="LogWriter")
@@ -151,27 +154,28 @@ class NonBlockingFileHandler:
         """
         获取或创建RotatingFileHandler实例
         """
-        if file_path not in self._rotating_handlers:
-            # 确保目录存在
-            file_path.parent.mkdir(parents=True, exist_ok=True)
+        with self._handlers_lock:
+            if file_path not in self._rotating_handlers:
+                # 确保目录存在
+                file_path.parent.mkdir(parents=True, exist_ok=True)
 
-            # 创建RotatingFileHandler
-            handler = RotatingFileHandler(
-                filename=str(file_path),
-                maxBytes=log_settings.LOG_MAX_FILE_SIZE_BYTES,
-                backupCount=log_settings.LOG_BACKUP_COUNT,
-                encoding='utf-8'
-            )
+                # 创建RotatingFileHandler
+                handler = RotatingFileHandler(
+                    filename=str(file_path),
+                    maxBytes=log_settings.LOG_MAX_FILE_SIZE_BYTES,
+                    backupCount=log_settings.LOG_BACKUP_COUNT,
+                    encoding='utf-8'
+                )
 
-            # 设置格式化器
-            formatter = logging.Formatter(log_settings.LOG_FILE_FORMAT)
-            handler.setFormatter(formatter)
+                # 设置格式化器
+                formatter = logging.Formatter(log_settings.LOG_FILE_FORMAT)
+                handler.setFormatter(formatter)
 
-            self._rotating_handlers[file_path] = handler
+                self._rotating_handlers[file_path] = handler
 
-        return self._rotating_handlers[file_path]
+            return self._rotating_handlers[file_path]
 
-    def write_log(self, level: str, message: str, file_path: Path):
+    def write_log(self, level: str, message: str, file_path: Path) -> None:
         """
         写入日志 - 自动检测协程环境并使用合适的方式
         """
@@ -181,8 +185,11 @@ class NonBlockingFileHandler:
         if self._is_in_event_loop():
             # 在协程环境中，使用非阻塞方式
             self._write_non_blocking(entry)
-        else:
-            # 不在协程环境中，直接同步写入
+            return
+        with self._state_lock:
+            if not self._running:
+                return
+            # 不在协程环境中，持锁同步写入，避免关闭文件处理器时仍有写操作进行
             self._write_sync(entry)
 
     @staticmethod
@@ -196,15 +203,19 @@ class NonBlockingFileHandler:
         except RuntimeError:
             return False
 
-    def _write_non_blocking(self, entry: LogEntry):
+    def _write_non_blocking(self, entry: LogEntry) -> bool:
         """
         非阻塞写入（用于协程环境）
         """
-        try:
-            self._write_queue.put_nowait(entry)
-        except queue.Full:
-            # 队列满时，使用线程池处理
-            self._executor.submit(self._write_sync, entry)
+        with self._state_lock:
+            if not self._running:
+                return False
+            try:
+                self._write_queue.put_nowait(entry)
+            except queue.Full:
+                # 队列满时，使用线程池处理
+                self._executor.submit(self._write_sync, entry)
+            return True
 
     @staticmethod
     def _write_sync(entry: LogEntry):
@@ -215,8 +226,7 @@ class NonBlockingFileHandler:
             # 获取RotatingFileHandler实例
             handler = NonBlockingFileHandler()._get_rotating_handler(entry.file_path)
 
-            # 使用RotatingFileHandler的emit方法，只传递原始消息
-            handler.emit(logging.LogRecord(
+            handler.handle(logging.LogRecord(
                 name='',
                 level=getattr(logging, entry.level.upper(), logging.INFO),
                 pathname='',
@@ -235,22 +245,28 @@ class NonBlockingFileHandler:
         """
         后台批量写入线程
         """
-        while self._running:
+        while True:
             try:
                 # 收集一批日志条目
                 batch = []
+                should_stop = False
                 end_time = time.time() + log_settings.WRITE_TIMEOUT
 
                 while len(batch) < log_settings.BATCH_WRITE_SIZE and time.time() < end_time:
                     try:
                         remaining_time = max(0, end_time - time.time())
                         entry = self._write_queue.get(timeout=remaining_time)
+                        if entry is self._stop_sentinel:
+                            should_stop = True
+                            break
                         batch.append(entry)
                     except queue.Empty:
                         break
 
                 if batch:
                     self._write_batch(batch)
+                if should_stop:
+                    break
 
             except Exception as e:
                 print(f"批量写入线程错误: {e}")
@@ -275,8 +291,7 @@ class NonBlockingFileHandler:
 
                 # 批量写入
                 for entry in entries:
-                    # 使用RotatingFileHandler的emit方法，只传递原始消息
-                    handler.emit(logging.LogRecord(
+                    handler.handle(logging.LogRecord(
                         name='',
                         level=getattr(logging, entry.level.upper(), logging.INFO),
                         pathname='',
@@ -294,15 +309,23 @@ class NonBlockingFileHandler:
 
     def shutdown(self):
         """
-        关闭文件处理器
+        排空异步日志并关闭文件处理器
         """
-        self._running = False
-        if hasattr(self, '_write_thread'):
-            self._write_thread.join(timeout=5)
+        with self._state_lock:
+            if not self._running:
+                return
+            self._running = False
+            if hasattr(self, '_write_thread') and self._write_thread.is_alive():
+                # 状态锁保证停止标记之后不会再有生产者入队
+                self._write_queue.put(self._stop_sentinel)
+        if hasattr(self, '_write_thread') and self._write_thread.is_alive():
+            self._write_thread.join()
         if self._executor:
             self._executor.shutdown(wait=True)
 
-        # 清理缓存
+        for handler in self._rotating_handlers.values():
+            handler.flush()
+            handler.close()
         self._rotating_handlers.clear()
 
 
diff --git a/app/modules/telegram/telegram.py b/app/modules/telegram/telegram.py
index 4d1b2b40..5eefb533 100644
--- a/app/modules/telegram/telegram.py
+++ b/app/modules/telegram/telegram.py
@@ -279,8 +279,6 @@ class Telegram:
     @staticmethod
     def _telegramify_item_text(item: Text) -> str:
         """将 telegramify 文本片段转换为 Telegram MarkdownV2 字符串。"""
-        if hasattr(item, "content"):
-            return item.content
         if entities_to_markdownv2:
             return entities_to_markdownv2(item.text, item.entities)
         return standardize(item.text)
@@ -290,8 +288,6 @@ class Telegram:
         """将 telegramify 文本或媒体片段转换为 Telegram MarkdownV2 caption。"""
         if isinstance(item, Text):
             return Telegram._telegramify_item_text(item)
-        if hasattr(item, "caption"):
-            return item.caption
         if entities_to_markdownv2:
             return entities_to_markdownv2(item.caption_text, item.caption_entities)
         return standardize(item.caption_text)
diff --git a/app/startup/lifecycle.py b/app/startup/lifecycle.py
index 0ac7f7d1..a40d1525 100644
--- a/app/startup/lifecycle.py
+++ b/app/startup/lifecycle.py
@@ -20,6 +20,7 @@ from app.chain.system import SystemChain
 from app.core.config import global_vars, settings
 from app.helper.server import MoviePilotServerHelper
 from app.helper.system import SystemHelper
+from app.log import LoggerManager
 from app.startup.command_initializer import init_command, stop_command, restart_command
 from app.startup.modules_initializer import init_modules, stop_modules
 from app.startup.monitor_initializer import stop_monitor, init_monitor
@@ -97,20 +98,24 @@ async def lifespan(app: FastAPI):
             pass
         except Exception as e:
             print(str(e))
-        if not settings.MOVIEPILOT_SAFE_MODE:
-            # 备份插件
-            SystemChain().backup_plugins()
-            # 停止工作流
-            stop_workflow()
-            # 停止命令
-            stop_command()
-            # 停止监控器
-            stop_monitor()
-            # 停止定时器
-            stop_scheduler()
-            # 停止插件
-            stop_plugins()
-        # 停止模块
-        await stop_modules()
-        # 关闭共享的异步 HTTP 连接池，释放底层连接资源
-        await aclose_shared_async_transports()
+        try:
+            if not settings.MOVIEPILOT_SAFE_MODE:
+                # 备份插件
+                SystemChain().backup_plugins()
+                # 停止工作流
+                stop_workflow()
+                # 停止命令
+                stop_command()
+                # 停止监控器
+                stop_monitor()
+                # 停止定时器
+                stop_scheduler()
+                # 停止插件
+                stop_plugins()
+            # 停止模块
+            await stop_modules()
+            # 关闭共享的异步 HTTP 连接池，释放底层连接资源
+            await aclose_shared_async_transports()
+        finally:
+            # 日志最后关闭，确保其他组件的收尾信息已写入文件
+            LoggerManager.shutdown()
diff --git a/app/startup/modules_initializer.py b/app/startup/modules_initializer.py
index c66ff5d2..840cfd3d 100644
--- a/app/startup/modules_initializer.py
+++ b/app/startup/modules_initializer.py
@@ -137,6 +137,8 @@ async def stop_modules():
     EventManager().stop()
     # 停止虚拟显示
     DisplayHelper().stop()
+    # 停止 DoH 服务
+    DohHelper().shutdown()
     # 停止线程池
     ThreadHelper().shutdown()
     # 停止消息服务
diff --git a/app/testing/network_guard.py b/app/testing/network_guard.py
index 9729e960..5e510cb7 100644
--- a/app/testing/network_guard.py
+++ b/app/testing/network_guard.py
@@ -9,6 +9,8 @@ fixture 一并识别，autouse 自动作用于每个用例，无需逐用例改
 """
 from __future__ import annotations
 
+import ipaddress
+
 import pytest
 
 # 本地回环/通配地址放行，其余主机一律视为真实出站；getaddrinfo 的 host 可能为 str 或 bytes
@@ -20,21 +22,45 @@ def block_real_network(monkeypatch):
     """防御纵深：拦截对非本地主机的真实出站，强制测试零真实网络。
 
     补在各用例自身 mock 之上：某用例万一漏 mock 外部依赖（TMDB / LLM 目录 / 下载器 /
-    媒体服务器 / 任意外链），其真实 DNS 解析会在此被拦并报错，而非静默发请求。本地回环放行
-    （sqlite 等）。asyncio 默认解析器经线程池调用 ``socket.getaddrinfo``，故拦此一处即覆盖
-    同步与异步出站。``monkeypatch`` 在用例结束后自动还原，不影响其他用例与进程退出。
+    媒体服务器 / 任意外链），其 DNS 解析或 socket 连接会被拦截。本地回环放行（sqlite 等）。
+    所有拦截记录会在用例收尾再次断言，避免业务代码捕获网络异常后让漏 mock 的用例静默通过。
+    ``monkeypatch`` 在用例结束后自动还原，不影响其他用例与进程退出。
     """
     import socket
 
     _real_getaddrinfo = socket.getaddrinfo
+    _real_connect = socket.socket.connect
+    attempts = []
+
+    def _is_allowed_host(host) -> bool:
+        normalized = host.decode() if isinstance(host, (bytes, bytearray)) else host
+        if normalized is None or normalized in _ALLOWED_NETWORK_HOSTS:
+            return True
+        try:
+            address = ipaddress.ip_address(str(normalized).split("%", 1)[0])
+            return address.is_loopback or address.is_unspecified
+        except ValueError:
+            return False
+
+    def _blocked(operation: str, host):
+        attempts.append((operation, host))
+        raise RuntimeError(
+            f"测试禁止真实出站网络：尝试通过 {operation} 访问 {host!r}；请 mock 对应外部依赖"
+        )
 
     def _guarded_getaddrinfo(host, *args, **kwargs):
-        normalized = host.decode() if isinstance(host, (bytes, bytearray)) else host
-        if normalized is not None and normalized not in _ALLOWED_NETWORK_HOSTS:
-            raise RuntimeError(
-                f"测试禁止真实出站网络：尝试解析 {normalized!r}；请 mock 对应外部依赖"
-            )
+        if not _is_allowed_host(host):
+            _blocked("DNS", host)
         return _real_getaddrinfo(host, *args, **kwargs)
 
+    def _guarded_connect(sock, address):
+        if isinstance(address, tuple) and address and not _is_allowed_host(address[0]):
+            _blocked("socket", address[0])
+        return _real_connect(sock, address)
+
     monkeypatch.setattr(socket, "getaddrinfo", _guarded_getaddrinfo)
+    monkeypatch.setattr(socket.socket, "connect", _guarded_connect)
     yield
+    if attempts:
+        details = ", ".join(f"{operation}:{host}" for operation, host in attempts)
+        pytest.fail(f"测试期间发生真实出站网络尝试：{details}")
diff --git a/app/utils/singleton.py b/app/utils/singleton.py
index 15503ab3..d98b9764 100644
--- a/app/utils/singleton.py
+++ b/app/utils/singleton.py
@@ -10,6 +10,11 @@ class Singleton(abc.ABCMeta, type):
 
     _instances: dict = {}
 
+    def get_existing_instance(cls, *args, **kwargs):
+        """按相同参数返回已创建实例，不触发初始化"""
+        key = (cls, args, frozenset(kwargs.items()))
+        return cls._instances.get(key)
+
     def __call__(cls, *args, **kwargs):
         key = (cls, args, frozenset(kwargs.items()))
         if key not in cls._instances:
@@ -31,6 +36,10 @@ class SingletonClass(abc.ABCMeta, type):
 
     _instances: dict = {}
 
+    def get_existing_instance(cls):
+        """返回已创建实例，不触发初始化"""
+        return cls._instances.get(cls)
+
     def __call__(cls, *args, **kwargs):
         if cls not in cls._instances:
             cls._instances[cls] = super().__call__(*args, **kwargs)
diff --git a/docker/entrypoint.sh b/docker/entrypoint.sh
index 5398c76a..fea389d1 100644
--- a/docker/entrypoint.sh
+++ b/docker/entrypoint.sh
@@ -20,6 +20,8 @@ function WARN() {
     echo -e "${WARN} ${1}"
 }
 
+ENTRYPOINT_START_TIME="$(date +%s)"
+
 function normalize_env_value() {
     printf '%s' "${1:-}" | tr '[:upper:]' '[:lower:]'
 }
@@ -57,6 +59,42 @@ function run_package_command() {
     fi
 }
 
+function wait_backend_ready() {
+    local entrypoint_start_time="${1:-$(date +%s)}"
+    local backend_start_time="${2:-$(date +%s)}"
+    local python_pid="${3:-}"
+    local backend_port="${PORT:-3001}"
+    local web_port="${NGINX_PORT:-3000}"
+    local timeout="${MOVIEPILOT_BACKEND_READY_TIMEOUT:-300}"
+    local ready_url="http://127.0.0.1:${backend_port}/api/v1/system/global?token=moviepilot"
+    local deadline
+    if ! [[ "${timeout}" =~ ^[0-9]+$ ]] || [ "$((10#${timeout}))" -le 0 ]; then
+        WARN "→ MOVIEPILOT_BACKEND_READY_TIMEOUT=${timeout} 无效，使用默认 300 秒。"
+        timeout=300
+    else
+        timeout=$((10#${timeout}))
+    fi
+    deadline=$(( $(date +%s) + timeout ))
+
+    while [ "$(date +%s)" -lt "${deadline}" ]; do
+        if [ -n "${python_pid}" ] && ! kill -0 "${python_pid}" >/dev/null 2>&1; then
+            WARN "→ 后端服务启动完成探测已停止：后端进程已退出。"
+            return 1
+        fi
+
+        if curl -fsS --max-time 2 "${ready_url}" >/dev/null 2>&1; then
+            local now
+            now="$(date +%s)"
+            INFO "→ MoviePilot Web 已可访问，启动总耗时 $(( now - entrypoint_start_time )) 秒，后端就绪耗时 $(( now - backend_start_time )) 秒，后端端口 ${backend_port}，前端端口 ${web_port}。"
+            return 0
+        fi
+        sleep 1
+    done
+
+    WARN "→ 后端服务启动完成探测超时，已等待 ${timeout} 秒，后端端口 ${backend_port}，继续等待进程日志..."
+    return 1
+}
+
 # 环境变量补全
 # 优先级: 系统环境变量 -> .env 文件 (即使为空字符串) -> 预设默认值
 # 精准适配 Python 端 set_key (quote_mode="always", 单引号包裹, \' 转义)
@@ -480,12 +518,14 @@ umask "${UMASK}"
 
 # 启动后端服务
 INFO "→ 启动后端服务..."
+BACKEND_START_TIME="$(date +%s)"
 if [ "${START_NOGOSU:-false}" = "true" ]; then
     "${VENV_PATH}/bin/python3" app/main.py > /dev/stdout 2> /dev/stderr &
 else
     gosu moviepilot:moviepilot "${VENV_PATH}/bin/python3" app/main.py > /dev/stdout 2> /dev/stderr &
 fi
 PYTHON_PID=$!
+wait_backend_ready "${ENTRYPOINT_START_TIME}" "${BACKEND_START_TIME}" "${PYTHON_PID}" &
 
 # 等待 Python 进程退出。
 # 如果收到信号，trap 会中断 wait，并执行 graceful_exit。
diff --git a/docs/pr-agent.md b/docs/pr-agent.md
index 646252f4..a13def77 100644
--- a/docs/pr-agent.md
+++ b/docs/pr-agent.md
@@ -1,86 +1,50 @@
 # PR-Agent 使用说明
 
-本仓库通过 GitHub Actions 运行开源 PR-Agent，用于自动维护 PR 摘要、发布行内代码审查建议，并在每轮审查后发布一条简短的 Code Review 总结评论。
+本仓库通过 GitHub Actions 运行 PR-Agent，帮助贡献者维护 PR 摘要、获取代码审查结果和提出 PR 相关问题。
 
-## 触发方式
+## 自动执行
 
-`.github/workflows/pr-agent.yml` 监听：
+同仓分支和来自 fork 的 PR 都会自动执行 PR-Agent。
 
-- `pull_request_target`：PR 打开、重新打开、标记 ready、请求 review、推送新 commit 时自动运行。
-- `issue_comment`：允许身份在 PR 评论里写允许的命令时手动运行。
+PR 在以下场景会自动处理：
 
-PR 事件会自动执行受控审查，包含同仓 PR 和 fork PR。允许身份也可以在 PR 评论中使用允许的命令触发受控审查。
-允许身份包括 `OWNER`、`MEMBER`、`COLLABORATOR`、`CONTRIBUTOR` 和 `FIRST_TIME_CONTRIBUTOR`。
+- 打开或重新打开 PR。
+- 将草稿 PR 标记为可审查。
+- 请求审查。
+- 每次推送新的 commit。
 
-## Workflow 权限
+PR 带有 `skip pr-agent` 标签，或标题以 `[Auto]`、`Auto` 开头时，自动和手工路径都会跳过。
 
-workflow 设置了最小可用权限：
+## 手工命令
 
-- `contents: read`：读取仓库内容和 PR diff。
-- `pull-requests: write`：更新 PR 描述、发布 PR Review 或修改 PR 相关元数据。
-- `issues: write`：PR 评论在 GitHub API 中属于 issue comments，手动命令和总结评论需要该权限。
-
-没有开启 `contents: write`。当前配置不让 PR-Agent 往仓库推代码或提交 changelog，因此不需要内容写权限。
-
-## 自动行为
-
-PR 事件默认自动执行：
-
-- `/describe`：更新 PR Body 中的 `PR-Agent 摘要` / `PR-Agent Summary` 标记区域，保留用户原始描述。
-- `/improve`：发布 GitHub 行内代码审查建议，不发布 PR-Agent 建议表格。
-
-workflow 会在 `/improve` 后发布一条普通 PR 评论：
-
-- 评论标题为 `## Code Review`。
-- 如果本轮有新增行内建议，会基于这些建议生成自然语言总结。
-- 如果本轮没有新增行内建议，直接发布无更多反馈的简短总结。
-- 下一次运行前会删除上一条 PR-Agent Code Review 总结评论，避免评论堆叠，同时保留新的通知事件。
-
-## 常用评论命令
-
-以下身份可在 PR 评论中使用：
-
-- `OWNER`：仓库所有者。
-- `MEMBER`：组织仓库中的组织成员。
-- `COLLABORATOR`：仓库协作者。
-- `CONTRIBUTOR`：曾经向仓库提交并合入过代码的贡献者。
-- `FIRST_TIME_CONTRIBUTOR`：首次向仓库贡献 PR 的用户。
+在 PR 的普通讨论评论中使用以下命令：
 
 ```text
 /describe
-/improve
+/review
 /ask 这次改动有没有遗漏权限校验？
 ```
 
-评论触发依赖 `issue_comment` 事件。普通 issue 评论、Bot 评论、非允许身份评论、以及不以允许命令开头的评论都会跳过。
+- `/describe`：更新 PR Body 内按语言显示的 `PR-Agent 摘要` 或 `PR-Agent Summary`，并保留贡献者原有的 PR 描述。
+- `/review`：发起一次代码审查。
+- `/ask ...`：就当前 PR 提问，回复会发布在普通 PR 评论中。
 
-## 输出约定
+手工命令仅允许以下 GitHub 身份关联的用户使用：`OWNER`、`MEMBER`、`COLLABORATOR`、`CONTRIBUTOR`、`FIRST_TIME_CONTRIBUTOR`。
 
-PR-Agent 配置集中在 `.github/workflows/pr-agent.yml` 中维护。公开说明只描述用户可见行为：
+新建的合法命令评论会触发执行；编辑后仍为合法命令的评论也会触发。编辑普通讨论评论不会调用模型。
 
-- 根据 PR 标题和用户原始描述自动选择中文或英文；无法识别时默认中文。
-- 保留用户原始 PR 描述，只更新 PR Body 中的 PR-Agent 标记区域。
-- 不使用 PR-Agent 的 Reviewer Guide 输出。
-- 不输出 PR Type、额外标签、图表或 describe 评论。
-- 只发布值得维护者处理的问题型行内建议。
-- 行内建议可使用 GitHub suggestion 形式，便于直接采纳。
-- 没有建议时不发布 PR-Agent 建议表格，只保留简短的 Code Review 总结评论。
+## 审查结果
 
-行内建议可使用风险前缀：
+`/describe` 的结果位于 PR Body 中按语言显示的 `PR-Agent 摘要` 或 `PR-Agent Summary` 区域，用于概览本次变更。
 
-- `🔴 **High Risk**:`：高风险问题。
-- `🟡 **Medium Risk**:`：中风险问题。
-- `🔵 **Low Risk**:`：低风险问题。
+`/review` 和自动审查会通过原生 GitHub Review 发布，结果位于 Review 页签，标题固定为 `PR-Agent Code Review`。审查摘要包含可点击的 `文件:行号` 链接；可定位到本次变更的具体问题会在对应代码行以行内评论呈现，无法行内定位的问题仍通过摘要中的链接呈现。
 
-可按需再启用的工具配置：
+审查不会额外创建专用的摘要评论。未发现需要处理的问题时，Review 会显示：
 
-- `[pr_update_changelog]`：配合 `/update_changelog` 生成 changelog 建议。
-- `[pr_add_docs]`：配合 `/add_docs` 生成文档建议。
-- `[pr_test]`：配合 `/test` 生成测试建议；它不会替代仓库自己的测试命令。
-- `[pr_questions]`：配合 `/ask ...` 回答 PR 相关问题。
+> 本次变更无需提出审查意见，暂无其他反馈。
 
 ## 安全边界
 
-PR-Agent 依赖的 Docker 镜像在 workflow 中固定版本号和 digest，不使用浮动的 `latest` 或仅依赖可变 tag。
+workflow 使用固定 digest 的 PR-Agent 容器镜像，不使用浮动标签。自动审查通过 `pull_request_target` 在目标仓库上下文中读取 PR 信息，但不会 checkout 或执行 PR 分支代码。
 
-当前使用 `pull_request_target` 支持 PR 自动审查，但 workflow 不 checkout 或执行 PR 分支代码，只运行固定 digest 的 PR-Agent 容器并通过 GitHub API 读取 PR diff。`issue_comment` 属于 base repo 事件，因此评论命令只允许指定身份触发。
+权限保持最小化：只授予读取仓库内容所需的 `contents: read`，以及更新 PR Body、发布 Review 和回复 PR 评论所需的写权限；不会向仓库推送代码或创建提交。
diff --git a/pytest.ini b/pytest.ini
index 85c4473a..4ec1fb1b 100644
--- a/pytest.ini
+++ b/pytest.ini
@@ -7,9 +7,5 @@ timeout_method = thread
 # 让本仓自身的新告警更醒目。本仓代码引发的告警一律不在此忽略，应在源码/用例处修复。
 filterwarnings =
     ignore:datetime.datetime.utcfromtimestamp\(\) is deprecated:DeprecationWarning
-    ignore:websockets.legacy is deprecated:DeprecationWarning
-    ignore:websockets.InvalidStatusCode is deprecated:DeprecationWarning
-    ignore:pkg_resources is deprecated as an API:DeprecationWarning
-    ignore:Deprecated call to .pkg_resources.declare_namespace:DeprecationWarning
     ignore:'crypt' is deprecated:DeprecationWarning
     ignore:'audioop' is deprecated:DeprecationWarning
diff --git a/tests/conftest.py b/tests/conftest.py
index 78a67f4d..573bf973 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -16,9 +16,11 @@ prepare_backend()
 from app.testing.network_guard import block_real_network  # noqa: E402,F401
 
 
-def _report_session_cleanup_error(name: str, err: Exception) -> None:
-    """测试收尾清理失败只记录诊断，不覆盖原始 pytest 退出状态。"""
+def _report_session_cleanup_error(session, name: str, err: Exception) -> None:
+    """记录收尾错误；原测试绿色时将会话标记为失败。"""
     sys.stderr.write(f"\npytest session cleanup failed: {name}: {err!r}\n")
+    if session.exitstatus == 0:
+        session.exitstatus = 1
 
 
 def pytest_sessionfinish(session, exitstatus):
@@ -28,21 +30,27 @@ def pytest_sessionfinish(session, exitstatus):
 
         shutdown_blocking_executors(cancel_futures=True)
     except Exception as err:
-        _report_session_cleanup_error("agent blocking executors", err)
+        _report_session_cleanup_error(session, "agent blocking executors", err)
 
     try:
         from app.helper.thread import ThreadHelper
-        from app.utils.singleton import Singleton
 
-        helper = Singleton._instances.get((ThreadHelper, (), frozenset()))
+        helper = ThreadHelper.get_existing_instance()
         if helper:
             helper.shutdown()
     except Exception as err:
-        _report_session_cleanup_error("thread helper", err)
+        _report_session_cleanup_error(session, "thread helper", err)
+
+    try:
+        from app.helper.message import stop_message
+
+        stop_message()
+    except Exception as err:
+        _report_session_cleanup_error(session, "message service", err)
 
     try:
         from app.log import LoggerManager
 
         LoggerManager.shutdown()
     except Exception as err:
-        _report_session_cleanup_error("logger manager", err)
+        _report_session_cleanup_error(session, "logger manager", err)
diff --git a/tests/test_db_error_diagnostics.py b/tests/test_db_error_diagnostics.py
new file mode 100644
index 00000000..8614237e
--- /dev/null
+++ b/tests/test_db_error_diagnostics.py
@@ -0,0 +1,106 @@
+import asyncio
+
+import pytest
+from sqlalchemy import create_engine, text
+from sqlalchemy.exc import OperationalError
+
+import app.db as db_module
+
+
+class _SqliteError(Exception):
+    """模拟 sqlite3 异常暴露的扩展错误字段。"""
+
+    sqlite_errorcode = 266
+    sqlite_errorname = "SQLITE_IOERR_READ"
+
+
+class _PsycopgError(Exception):
+    """模拟 psycopg2 异常暴露的 SQLSTATE 字段。"""
+
+    pgcode = "40001"
+
+
+class _AsyncpgError(Exception):
+    """模拟 asyncpg 适配异常暴露的 SQLSTATE 字段。"""
+
+    sqlstate = "23505"
+
+
+@pytest.mark.parametrize(
+    ("error", "expected"),
+    [
+        (
+            _SqliteError("disk I/O error"),
+            {
+                "error_type": "_SqliteError",
+                "error_code": 266,
+                "error_name": "SQLITE_IOERR_READ",
+            },
+        ),
+        (
+            _PsycopgError("serialization failure"),
+            {
+                "error_type": "_PsycopgError",
+                "sqlstate": "40001",
+            },
+        ),
+        (
+            _AsyncpgError("duplicate key"),
+            {
+                "error_type": "_AsyncpgError",
+                "sqlstate": "23505",
+            },
+        ),
+    ],
+)
+def test_database_error_metadata_extracts_driver_codes(error, expected) -> None:
+    """诊断元数据应兼容 SQLite、psycopg2 与 asyncpg 的稳定错误字段。"""
+    assert db_module._database_error_metadata(error) == expected
+
+
+def test_database_error_listener_omits_statement_and_parameters(monkeypatch) -> None:
+    """数据库错误日志不得包含 SQL、参数或驱动返回的原始消息。"""
+    messages = []
+    engine = create_engine("sqlite:///:memory:")
+    monkeypatch.setattr("app.db.logger.error", messages.append)
+    db_module._register_database_error_logging(engine)
+
+    with pytest.raises(OperationalError):
+        with engine.connect() as connection:
+            connection.execute(
+                text("SELECT * FROM missing_table WHERE token = :token"),
+                {"token": "private-token"},
+            )
+
+    assert len(messages) == 1
+    assert "database=sqlite" in messages[0]
+    assert "driver=pysqlite" in messages[0]
+    assert "error_code=1" in messages[0]
+    assert "error_name=SQLITE_ERROR" in messages[0]
+    assert "missing_table" not in messages[0]
+    assert "private-token" not in messages[0]
+
+
+def test_async_database_engine_logs_driver_error_metadata(monkeypatch) -> None:
+    """异步 Engine 应通过底层 sync engine 记录驱动错误码。"""
+    messages = []
+    monkeypatch.setattr("app.db.logger.error", messages.append)
+
+    async def query_missing_table() -> None:
+        async with db_module.AsyncEngine.connect() as connection:
+            await connection.execute(text("SELECT * FROM async_missing_table"))
+
+    with pytest.raises(OperationalError):
+        asyncio.run(query_missing_table())
+
+    assert len(messages) == 1
+    assert "database=sqlite" in messages[0]
+    assert "driver=aiosqlite" in messages[0]
+    assert "error_code=1" in messages[0]
+    assert "error_name=SQLITE_ERROR" in messages[0]
+    assert "async_missing_table" not in messages[0]
+
+
+def test_database_error_metadata_ignores_unclassified_errors() -> None:
+    """没有驱动错误码时不应制造无效诊断日志。"""
+    assert db_module._database_error_metadata(RuntimeError("plain failure")) is None
diff --git a/tests/test_docker_entrypoint_permissions.py b/tests/test_docker_entrypoint_permissions.py
index 2ae66557..f5862ef6 100644
--- a/tests/test_docker_entrypoint_permissions.py
+++ b/tests/test_docker_entrypoint_permissions.py
@@ -78,6 +78,26 @@ def _run_permission_case(tmp_path: Path, body: str, env: dict[str, str] | None =
     return chown_log.read_text(encoding="utf-8") if chown_log.exists() else ""
 
 
+def _run_entrypoint_case(tmp_path: Path, body: str, env: dict[str, str] | None = None) -> str:
+    functions = _write_entrypoint_functions(tmp_path)
+    case_env = {
+        **os.environ,
+        "ENTRYPOINT_FUNCTIONS": str(functions),
+    }
+    if env:
+        case_env.update(env)
+
+    script = textwrap.dedent(
+        f"""\
+        set -euo pipefail
+        source "${{ENTRYPOINT_FUNCTIONS}}"
+        {body}
+        """
+    )
+    result = subprocess.run(["bash", "-c", script], check=True, env=case_env, text=True, capture_output=True)
+    return result.stdout
+
+
 def test_image_paths_are_not_chowned_by_default_regardless_of_owner(tmp_path: Path) -> None:
     log = _run_permission_case(
         tmp_path,
@@ -170,3 +190,56 @@ def test_runtime_writable_paths_are_still_corrected(tmp_path: Path) -> None:
     assert not any(line.startswith("-R ") and ".cloakbrowser" in line for line in lines)
     assert not any(f"{tmp_path}/app " in line for line in lines)
     assert not any(f"{tmp_path}/public" in line for line in lines)
+
+
+def test_backend_ready_log_uses_configured_ports(tmp_path: Path) -> None:
+    curl_log = tmp_path / "curl.log"
+    output = _run_entrypoint_case(
+        tmp_path,
+        """
+        INFO() { printf '[INFO] %s\\n' "$1"; }
+        curl() {
+          printf '%s\\n' "$*" > "${CURL_LOG}"
+          return 0
+        }
+        PORT=4321 NGINX_PORT=8765 wait_backend_ready 1 2 "$$"
+        """,
+        env={"CURL_LOG": str(curl_log)},
+    )
+
+    assert curl_log.read_text(encoding="utf-8") == (
+        "-fsS --max-time 2 http://127.0.0.1:4321/api/v1/system/global?token=moviepilot\n"
+    )
+    assert "MoviePilot Web 已可访问" in output
+    assert "后端就绪耗时" in output
+    assert "后端端口 4321" in output
+    assert "前端端口 8765" in output
+
+
+def test_backend_ready_timeout_falls_back_to_default_for_invalid_value(tmp_path: Path) -> None:
+    output = _run_entrypoint_case(
+        tmp_path,
+        """
+        WARN() { printf '[WARN] %s\\n' "$1"; }
+        curl() { return 1; }
+        MOVIEPILOT_BACKEND_READY_TIMEOUT=invalid wait_backend_ready 1 2 999999 || true
+        """,
+    )
+
+    assert "MOVIEPILOT_BACKEND_READY_TIMEOUT=invalid 无效，使用默认 300 秒" in output
+    assert "后端服务启动完成探测已停止：后端进程已退出" in output
+
+
+def test_backend_ready_timeout_accepts_leading_zero_decimal(tmp_path: Path) -> None:
+    output = _run_entrypoint_case(
+        tmp_path,
+        """
+        INFO() { printf '[INFO] %s\\n' "$1"; }
+        WARN() { printf '[WARN] %s\\n' "$1"; }
+        curl() { return 0; }
+        MOVIEPILOT_BACKEND_READY_TIMEOUT=08 wait_backend_ready 1 2 "$$"
+        """,
+    )
+
+    assert "MOVIEPILOT_BACKEND_READY_TIMEOUT=08 无效" not in output
+    assert "MoviePilot Web 已可访问" in output
diff --git a/tests/test_doh_helper.py b/tests/test_doh_helper.py
index 544483c9..07c53469 100644
--- a/tests/test_doh_helper.py
+++ b/tests/test_doh_helper.py
@@ -3,6 +3,61 @@ import socket
 from app.helper import doh
 
 
+def test_doh_executor_is_lazy_and_shutdown_restores_socket(monkeypatch):
+    """DoH 线程池按需创建，并在模块关闭时恢复系统 DNS"""
+    original_getaddrinfo = socket.getaddrinfo
+    helper = object.__new__(doh.DohHelper)
+    monkeypatch.setattr(doh.settings, "DOH_DOMAINS", "example.com")
+    monkeypatch.setattr(doh.settings, "DOH_RESOLVERS", "resolver.test")
+    monkeypatch.setattr(doh, "_doh_query", lambda resolver, host: "203.0.113.7")
+    monkeypatch.setattr(doh, "_orig_getaddrinfo", lambda host, *args, **kwargs: [])
+
+    try:
+        helper.shutdown()
+        assert doh._executor is None
+
+        doh.enable_doh(True)
+        socket.getaddrinfo("example.com", None)
+        executor = doh._executor
+        assert executor is not None
+
+        helper.shutdown()
+
+        assert doh._executor is None
+        assert socket.getaddrinfo is doh._orig_getaddrinfo
+        assert getattr(executor, "_shutdown", False)
+    finally:
+        helper.shutdown()
+        socket.getaddrinfo = original_getaddrinfo
+
+
+def test_doh_config_reload_disables_and_closes_executor(monkeypatch):
+    """热更新关闭 DoH 时恢复系统 DNS 并释放已创建的线程池"""
+    original_getaddrinfo = socket.getaddrinfo
+    helper = object.__new__(doh.DohHelper)
+    monkeypatch.setattr(doh.settings, "DOH_DOMAINS", "example.com")
+    monkeypatch.setattr(doh.settings, "DOH_RESOLVERS", "resolver.test")
+    monkeypatch.setattr(doh, "_doh_query", lambda resolver, host: "203.0.113.7")
+    monkeypatch.setattr(doh, "_orig_getaddrinfo", lambda host, *args, **kwargs: [])
+
+    try:
+        helper.shutdown()
+        doh.enable_doh(True)
+        socket.getaddrinfo("example.com", None)
+        executor = doh._executor
+        assert executor is not None
+        monkeypatch.setattr(doh.settings, "DOH_ENABLE", False)
+
+        helper.on_config_changed()
+
+        assert doh._executor is None
+        assert getattr(executor, "_shutdown", False)
+        assert socket.getaddrinfo is doh._orig_getaddrinfo
+    finally:
+        helper.shutdown()
+        socket.getaddrinfo = original_getaddrinfo
+
+
 def test_enable_doh_reuses_cached_host_resolution(monkeypatch):
     """
     同一 DoH 域名第二次解析应命中缓存，避免重复请求远端解析器。
@@ -33,6 +88,7 @@ def test_enable_doh_reuses_cached_host_resolution(monkeypatch):
         socket.getaddrinfo("example.com", None)
         socket.getaddrinfo("example.com", None)
     finally:
+        object.__new__(doh.DohHelper).shutdown()
         socket.getaddrinfo = original_getaddrinfo
         with doh._doh_lock:
             doh._doh_cache.clear()
diff --git a/tests/test_emby_dashboard_links.py b/tests/test_emby_dashboard_links.py
index ba886cf7..1b65bdaa 100644
--- a/tests/test_emby_dashboard_links.py
+++ b/tests/test_emby_dashboard_links.py
@@ -102,6 +102,7 @@ class EmbyDashboardLinksTest(unittest.TestCase):
         with (
             patch.object(client, "_Emby__get_emby_librarys") as librarys,
             patch.object(client, "_Emby__get_local_image_by_id") as image_by_id,
+            patch.object(client, "get_items_count", return_value=0),
         ):
             librarys.return_value = [
                 {
diff --git a/tests/test_lifecycle_shutdown.py b/tests/test_lifecycle_shutdown.py
new file mode 100644
index 00000000..c984a3f5
--- /dev/null
+++ b/tests/test_lifecycle_shutdown.py
@@ -0,0 +1,46 @@
+import asyncio
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from fastapi import FastAPI
+
+from app.startup import lifecycle
+
+
+def test_lifespan_closes_logger_when_early_shutdown_step_fails(monkeypatch):
+    """前置关闭步骤失败时仍应关闭 Logger"""
+    monkeypatch.setattr(lifecycle.settings, "MOVIEPILOT_SAFE_MODE", False)
+    monkeypatch.setattr(lifecycle.global_vars, "set_loop", MagicMock())
+    for name in (
+        "init_routers",
+        "init_modules",
+        "init_plugins",
+        "init_scheduler",
+        "init_monitor",
+        "init_command",
+        "init_workflow",
+        "stop_workflow",
+        "stop_command",
+        "stop_monitor",
+        "stop_scheduler",
+        "stop_plugins",
+    ):
+        monkeypatch.setattr(lifecycle, name, MagicMock())
+
+    system_chain = MagicMock()
+    system_chain.backup_plugins.side_effect = RuntimeError("backup failed")
+    monkeypatch.setattr(lifecycle, "SystemChain", MagicMock(return_value=system_chain))
+    monkeypatch.setattr(lifecycle, "init_extra", AsyncMock())
+    monkeypatch.setattr(lifecycle, "stop_modules", AsyncMock())
+    monkeypatch.setattr(lifecycle, "aclose_shared_async_transports", AsyncMock())
+    logger_shutdown = MagicMock()
+    monkeypatch.setattr(lifecycle.LoggerManager, "shutdown", logger_shutdown)
+
+    async def run_lifespan():
+        with pytest.raises(RuntimeError, match="backup failed"):
+            async with lifecycle.lifespan(FastAPI()):
+                pass
+
+    asyncio.run(run_lifespan())
+
+    logger_shutdown.assert_called_once_with()
diff --git a/tests/test_log_shutdown.py b/tests/test_log_shutdown.py
new file mode 100644
index 00000000..8a2ddfbb
--- /dev/null
+++ b/tests/test_log_shutdown.py
@@ -0,0 +1,149 @@
+import threading
+import time
+from unittest.mock import MagicMock
+
+from app.log import LogEntry, NonBlockingFileHandler, log_settings
+
+
+def test_non_blocking_file_handler_shutdown_wakes_writer_and_closes_handlers(tmp_path):
+    """日志关闭应立即唤醒空闲写线程，并关闭所有已打开的文件处理器"""
+    original_instance = NonBlockingFileHandler._instance
+    NonBlockingFileHandler._instance = None
+    handler = NonBlockingFileHandler()
+    handler._rotating_handlers = {}
+    log_handler = handler._get_rotating_handler(tmp_path / "shutdown.log")
+
+    try:
+        started_at = time.monotonic()
+        handler.shutdown()
+        elapsed = time.monotonic() - started_at
+
+        assert elapsed < 1
+        assert not handler._write_thread.is_alive()
+        assert log_handler.stream is None
+        assert handler._write_non_blocking(
+            LogEntry("info", "late-message", tmp_path / "shutdown.log")
+        ) is False
+        assert handler._write_queue.empty()
+    finally:
+        if handler._write_thread.is_alive():
+            handler._running = False
+            handler._write_thread.join(timeout=5)
+        if log_handler.stream is not None:
+            log_handler.close()
+        NonBlockingFileHandler._instance = original_instance
+
+
+def test_non_blocking_file_handler_shutdown_drains_queued_batches(monkeypatch, tmp_path):
+    """停止标记之前已进入队列的日志应跨批次全部写完"""
+    original_instance = NonBlockingFileHandler._instance
+    NonBlockingFileHandler._instance = None
+    monkeypatch.setattr(log_settings, "BATCH_WRITE_SIZE", 2)
+    handler = NonBlockingFileHandler()
+    handler._rotating_handlers = {}
+    written = []
+    monkeypatch.setattr(
+        handler,
+        "_write_batch",
+        lambda batch: written.extend(entry.message for entry in batch),
+    )
+
+    try:
+        for index in range(5):
+            handler._write_non_blocking(
+                LogEntry("info", f"message-{index}", tmp_path / "drain.log")
+            )
+
+        handler.shutdown()
+
+        assert written == [f"message-{index}" for index in range(5)]
+        assert not handler._write_thread.is_alive()
+    finally:
+        if handler._write_thread.is_alive():
+            handler._running = False
+            handler._write_queue.put(handler._stop_sentinel)
+            handler._write_thread.join(timeout=5)
+        NonBlockingFileHandler._instance = original_instance
+
+
+def test_non_blocking_file_handler_creates_one_handler_for_concurrent_first_write(monkeypatch, tmp_path):
+    """同一路径首次并发写入时只创建并关闭一个文件处理器"""
+    original_instance = NonBlockingFileHandler._instance
+    NonBlockingFileHandler._instance = None
+    handler = NonBlockingFileHandler()
+    handler._rotating_handlers = {}
+    first_created = threading.Event()
+    second_started = threading.Event()
+    release_first = threading.Event()
+    created_handlers = []
+    results = []
+
+    class ProbeHandler:
+        def __init__(self, **kwargs):
+            self.closed = False
+            created_handlers.append(self)
+            if len(created_handlers) == 1:
+                first_created.set()
+                release_first.wait(timeout=2)
+
+        @staticmethod
+        def setFormatter(formatter):
+            pass
+
+        @staticmethod
+        def flush():
+            pass
+
+        def close(self):
+            self.closed = True
+
+    monkeypatch.setattr("app.log.RotatingFileHandler", ProbeHandler)
+    file_path = tmp_path / "concurrent.log"
+
+    def get_handler(started=None):
+        if started:
+            started.set()
+        results.append(handler._get_rotating_handler(file_path))
+
+    first = threading.Thread(target=get_handler)
+    second = threading.Thread(target=get_handler, args=(second_started,))
+    try:
+        first.start()
+        assert first_created.wait(timeout=1)
+        second.start()
+        assert second_started.wait(timeout=1)
+        time.sleep(0.05)
+        release_first.set()
+        first.join(timeout=2)
+        second.join(timeout=2)
+
+        assert len(created_handlers) == 1
+        assert results[0] is results[1]
+
+        handler.shutdown()
+        assert created_handlers[0].closed is True
+    finally:
+        release_first.set()
+        first.join(timeout=2)
+        second.join(timeout=2)
+        handler.shutdown()
+        NonBlockingFileHandler._instance = original_instance
+
+
+def test_non_blocking_file_handler_uses_handler_lock(monkeypatch, tmp_path):
+    """日志写入通过 Handler 入口串行化 emit 与 rollover"""
+    original_instance = NonBlockingFileHandler._instance
+    NonBlockingFileHandler._instance = None
+    handler = NonBlockingFileHandler()
+    handler._rotating_handlers = {}
+    log_handler = MagicMock()
+    monkeypatch.setattr(handler, "_get_rotating_handler", MagicMock(return_value=log_handler))
+
+    try:
+        handler._write_sync(LogEntry("info", "message", tmp_path / "locked.log"))
+
+        log_handler.handle.assert_called_once()
+        log_handler.emit.assert_not_called()
+    finally:
+        handler.shutdown()
+        NonBlockingFileHandler._instance = original_instance
diff --git a/tests/test_media_interaction.py b/tests/test_media_interaction.py
index c777ce66..3648ea31 100644
--- a/tests/test_media_interaction.py
+++ b/tests/test_media_interaction.py
@@ -20,6 +20,16 @@ def clear_media_interactions():
     plugin_input_interaction_manager.clear()
 
 
+@pytest.fixture(autouse=True)
+def mock_default_media_search():
+    """未显式验证搜索结果的消息路由用例不访问真实媒体元数据服务"""
+    with patch(
+        "app.chain.media.MediaChain.search",
+        side_effect=lambda title: (_build_meta(title), []),
+    ):
+        yield
+
+
 def _build_meta(name: str) -> MetaBase:
     """构造媒体识别元数据。"""
     meta = MetaBase(name)
diff --git a/tests/test_message_queue_shutdown.py b/tests/test_message_queue_shutdown.py
new file mode 100644
index 00000000..9d37a1b2
--- /dev/null
+++ b/tests/test_message_queue_shutdown.py
@@ -0,0 +1,30 @@
+import time
+
+from app.helper.message import MessageQueueManager, TemplateHelper, stop_message
+from app.utils.singleton import SingletonClass
+
+
+def test_message_queue_stop_wakes_idle_monitor(monkeypatch):
+    """消息队列停止时应唤醒空闲监控线程，不等待完整检查周期"""
+    monkeypatch.setattr(MessageQueueManager, "init_config", lambda self: None)
+    manager = object.__new__(MessageQueueManager)
+    manager.__init__(check_interval=10)
+
+    started_at = time.monotonic()
+    manager.stop()
+    elapsed = time.monotonic() - started_at
+
+    assert elapsed < 1
+    assert not manager.thread.is_alive()
+
+
+def test_stop_message_does_not_initialize_absent_services(monkeypatch):
+    """消息服务未初始化时，关闭入口不应为了清理而创建后台资源"""
+    monkeypatch.setattr(SingletonClass, "_instances", {})
+
+    assert MessageQueueManager.get_existing_instance() is None
+    assert TemplateHelper.get_existing_instance() is None
+    stop_message()
+
+    assert MessageQueueManager not in SingletonClass._instances
+    assert TemplateHelper not in SingletonClass._instances
diff --git a/tests/test_network_guard.py b/tests/test_network_guard.py
new file mode 100644
index 00000000..cc1ed752
--- /dev/null
+++ b/tests/test_network_guard.py
@@ -0,0 +1,22 @@
+import socket
+
+import pytest
+
+from app.testing.network_guard import block_real_network
+
+
+def test_network_guard_fails_when_blocked_attempt_is_swallowed(monkeypatch):
+    """业务代码即使捕获网络异常，网络守卫仍应在用例收尾报告失败"""
+    fixture = block_real_network.__wrapped__(monkeypatch)
+    next(fixture)
+
+    try:
+        try:
+            socket.getaddrinfo("external.example", 443)
+        except RuntimeError:
+            pass
+
+        with pytest.raises(pytest.fail.Exception, match="external.example"):
+            next(fixture)
+    finally:
+        monkeypatch.undo()
diff --git a/tests/test_singleton.py b/tests/test_singleton.py
new file mode 100644
index 00000000..25e6e480
--- /dev/null
+++ b/tests/test_singleton.py
@@ -0,0 +1,29 @@
+from app.utils.singleton import Singleton, SingletonClass
+
+
+def test_singleton_class_can_read_existing_instance_without_creating(monkeypatch):
+    """按类单例可以只读取已存在实例"""
+
+    class Example(metaclass=SingletonClass):
+        pass
+
+    monkeypatch.setattr(SingletonClass, "_instances", {})
+
+    assert Example.get_existing_instance() is None
+    instance = Example()
+    assert Example.get_existing_instance() is instance
+
+
+def test_parameterized_singleton_can_read_matching_instance_without_creating(monkeypatch):
+    """参数化单例按相同参数读取已存在实例"""
+
+    class Example(metaclass=Singleton):
+        def __init__(self, name):
+            self.name = name
+
+    monkeypatch.setattr(Singleton, "_instances", {})
+
+    assert Example.get_existing_instance("first") is None
+    instance = Example("first")
+    assert Example.get_existing_instance("first") is instance
+    assert Example.get_existing_instance("second") is None
diff --git a/tests/test_subscribe_endpoint.py b/tests/test_subscribe_endpoint.py
index 2e34d8b5..ae1c0960 100644
--- a/tests/test_subscribe_endpoint.py
+++ b/tests/test_subscribe_endpoint.py
@@ -362,7 +362,7 @@ class SubscribeEndpointTest(TestCase):
         with patch(
             "app.api.endpoints.subscribe.Subscribe.async_list_by_username",
             new=AsyncMock(return_value=owned),
-        ):
+        ), patch("app.api.endpoints.subscribe.Scheduler") as scheduler_cls:
             response = asyncio.run(
                 search_subscribes(
                     background_tasks=background_tasks,
@@ -376,6 +376,7 @@ class SubscribeEndpointTest(TestCase):
             [task["kwargs"]["sid"] for task in background_tasks.tasks],
             [17, 18],
         )
+        self.assertEqual(scheduler_cls.return_value.start.call_count, 0)
 
     def test_subscribe_files_hides_other_user_row(self):
         """
diff --git a/tests/test_telegram.py b/tests/test_telegram.py
index 1d142d0e..6b9cecb6 100644
--- a/tests/test_telegram.py
+++ b/tests/test_telegram.py
@@ -3,6 +3,7 @@
 Telegram 模块单元测试（pytest 原生）。
 """
 import json
+import warnings
 from types import SimpleNamespace
 from unittest.mock import MagicMock, Mock, patch
 
@@ -286,13 +287,29 @@ def test_send_msg_markdown_escaping(telegram):
     assert send_kwargs["text"].startswith("*测试标题*\n")
 
 
-def test_telegramify_new_content_fields_are_used_directly():
-    """新版telegramify对象应直接使用已渲染的MarkdownV2字段"""
-    text_item = SimpleNamespace(content="已转义\\_文本")
-    file_item = SimpleNamespace(caption="已转义\\_说明")
+def test_telegramify_current_fields_are_used_directly():
+    """telegramify 对象直接使用当前 MarkdownV2 字段"""
+    from telegramify_markdown.content import ContentTrace, File, Text
 
-    assert Telegram._telegramify_item_text(text_item) == "已转义\\_文本"
-    assert Telegram._telegramify_item_caption(file_item) == "已转义\\_说明"
+    text_item = Text(
+        text="已转义_文本",
+        entities=[],
+        content_trace=ContentTrace(source_type="test"),
+    )
+    file_item = File(
+        file_name="test.txt",
+        file_data=b"test",
+        caption_text="已转义_说明",
+        caption_entities=[],
+        content_trace=ContentTrace(source_type="test"),
+    )
+
+    with warnings.catch_warnings(record=True) as warning_records:
+        warnings.simplefilter("always")
+        assert Telegram._telegramify_item_text(text_item) == "已转义\\_文本"
+        assert Telegram._telegramify_item_caption(file_item) == "已转义\\_说明"
+
+    assert not warning_records
 
 
 def test_send_msg_with_html_parse_mode_keeps_html(telegram):
diff --git a/tests/test_web_agent_stream.py b/tests/test_web_agent_stream.py
index f7d033c5..617bac33 100644
--- a/tests/test_web_agent_stream.py
+++ b/tests/test_web_agent_stream.py
@@ -183,7 +183,8 @@ def test_build_web_agent_command_items_returns_slash_commands():
 
 def test_build_web_agent_command_items_includes_sites_command():
     """WebAgent 命令建议应包含内建站点管理命令。"""
-    commands = _build_web_agent_command_items()
+    with patch("app.command.Scheduler"), patch("app.command.ThreadHelper"):
+        commands = _build_web_agent_command_items()
 
     assert any(command["command"] == "/sites" for command in commands)
 
diff --git a/version.py b/version.py
index 0d840a32..35c1b00c 100644
--- a/version.py
+++ b/version.py
@@ -1,2 +1,2 @@
-APP_VERSION = 'v2.14.2'
-FRONTEND_VERSION = 'v2.14.2'
+APP_VERSION = 'v2.14.3'
+FRONTEND_VERSION = 'v2.14.3'