fix(ci): 修复CI Gate失败时返回exit 0而非exit 1的P0 Bug

CI Gate判断失败时脚本错误地写了exit 0，导致所有PR不管CI失败成什么样， Gate永远显示success，分支保护完全失效。修复：将失败分支的exit 0改为exit 1
fix(ci): 格式修复防循环索引 + AI审查fail-open（2个bug修复） (#1045 )
2026-07-28 10:12:35 +08:00 · 2026-07-28 09:52:23 +08:00 · 2026-07-27 21:40:07 +08:00 · 2026-07-27 20:29:56 +08:00 · 2026-07-27 16:21:19 +08:00 · 2026-07-27 16:17:50 +08:00
8 changed files with 1246 additions and 195 deletions
@@ -36,12 +36,11 @@ jobs:
      GITEA_REPO: xiaoxia/xiaoxia-saas
    steps:
      - name: Checkout code
-        uses: actions/checkout@v4
-
-      - name: Setup Python
-        uses: actions/setup-python@v5
-        with:
-          python-version: '3.12'
+        shell: sh
+        env:
+          GITHUB_TOKEN: ${{ github.token }}
+        run: |
+          curl -sH "Authorization: token $GITHUB_TOKEN"             "${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/raw/scripts/ci/step_checkout.sh?ref=${GITHUB_SHA}"             | bash

      # ====== Cron模式：获取staging运行中镜像作为白名单 ======
      - name: Get staging running images (whitelist)
@@ -1662,3 +1662,160 @@ jobs:
        START_TIME=""
        [ -f /tmp/ci_job_start_time ] && START_TIME=$(cat /tmp/ci_job_start_time)
        python3 scripts/ci/ci_trace_report.py --service xiaoxia-saas-ci --status $STATUS --start-time "$START_TIME" || true
+
+  ci-gate:
+    name: CI Gate
+    runs-on: ci-l2
+    if: always() && github.event_name == 'pull_request'
+    needs:
+      - check-frontend-only
+      - validate-code-quality
+      - validate-type-check
+      - validate-migration
+      - unit-tests
+      - integration-tests
+      - frontend-lint
+      - frontend-unit-test
+      - build-pr
+    timeout-minutes: 3
+    steps:
+      - name: Checkout code
+        shell: sh
+        env:
+          GITHUB_TOKEN: ${{ github.token }}
+        run: |
+          curl -sH "Authorization: token $GITHUB_TOKEN" \
+            "${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/raw/scripts/ci/step_checkout.sh?ref=${GITHUB_SHA}" \
+            | bash
+
+      - name: Evaluate CI Gate
+        id: gate
+        shell: bash
+        env:
+          GITHUB_TOKEN: ${{ github.token }}
+          PR_NUMBER: ${{ github.event.pull_request.number }}
+          PR_HEAD_SHA: ${{ github.event.pull_request.head.sha }}
+          RESULT_CHECK_FRONTEND: ${{ needs.check-frontend-only.result }}
+          RESULT_CODE_QUALITY: ${{ needs.validate-code-quality.result }}
+          RESULT_TYPE_CHECK: ${{ needs.validate-type-check.result }}
+          RESULT_MIGRATION: ${{ needs.validate-migration.result }}
+          RESULT_UNIT_TESTS: ${{ needs.unit-tests.result }}
+          RESULT_INTEGRATION: ${{ needs.integration-tests.result }}
+          RESULT_FRONTEND_LINT: ${{ needs.frontend-lint.result }}
+          RESULT_FRONTEND_UNIT: ${{ needs.frontend-unit-test.result }}
+          RESULT_BUILD_PR: ${{ needs.build-pr.result }}
+        run: |
+          set -eu
+          echo "=== CI Gate 评估 ==="
+          echo ""
+          echo "各job结果:"
+          echo "  check-frontend-only: $RESULT_CHECK_FRONTEND"
+          echo "  validate-code-quality: $RESULT_CODE_QUALITY"
+          echo "  validate-type-check: $RESULT_TYPE_CHECK"
+          echo "  validate-migration: $RESULT_MIGRATION"
+          echo "  unit-tests: $RESULT_UNIT_TESTS"
+          echo "  integration-tests: $RESULT_INTEGRATION"
+          echo "  frontend-lint: $RESULT_FRONTEND_LINT"
+          echo "  frontend-unit-test: $RESULT_FRONTEND_UNIT"
+          echo "  build-pr: $RESULT_BUILD_PR"
+          echo ""
+
+          # 判断PR类型
+          SKIP_BACKEND="${{ needs.check-frontend-only.outputs.skip_backend }}"
+          SKIP_FRONTEND="${{ needs.check-frontend-only.outputs.skip_frontend }}"
+          echo "PR类型: skip_backend=$SKIP_BACKEND, skip_frontend=$SKIP_FRONTEND"
+
+          # 必填检查项（根据PR类型决定）
+          # 通用检查（所有PR都必须过）
+          REQUIRED_GENERAL=(
+            "validate-code-quality:$RESULT_CODE_QUALITY"
+            "validate-type-check:$RESULT_TYPE_CHECK"
+            "validate-migration:$RESULT_MIGRATION"
+            "frontend-lint:$RESULT_FRONTEND_LINT"
+            "build-pr:$RESULT_BUILD_PR"
+          )
+
+          # 后端检查
+          REQUIRED_BACKEND=(
+            "unit-tests:$RESULT_UNIT_TESTS"
+          )
+
+          # 前端检查
+          REQUIRED_FRONTEND=(
+            "frontend-unit-test:$RESULT_FRONTEND_UNIT"
+          )
+
+          ALL_PASSED=true
+          FAILED_ITEMS=()
+
+          check_job() {
+            local name=$1
+            local result=$2
+            if [ "$result" = "success" ]; then
+              echo "  ✅ $name: success"
+            elif [ "$result" = "skipped" ]; then
+              echo "  ⏭️  $name: skipped（跳过，不影响）"
+            else
+              echo "  ❌ $name: $result"
+              ALL_PASSED=false
+              FAILED_ITEMS+=("$name=$result")
+            fi
+          }
+
+          echo ""
+          echo "=== 通用检查（所有PR必填）==="
+          for item in "${REQUIRED_GENERAL[@]}"; do
+            name="${item%%:*}"
+            result="${item##*:}"
+            check_job "$name" "$result"
+          done
+
+          if [ "$SKIP_BACKEND" != "true" ]; then
+            echo ""
+            echo "=== 后端检查 ==="
+            for item in "${REQUIRED_BACKEND[@]}"; do
+              name="${item%%:*}"
+              result="${item##*:}"
+              check_job "$name" "$result"
+            done
+          else
+            echo ""
+            echo "=== 后端检查（纯前端PR，跳过）==="
+          fi
+
+          if [ "$SKIP_FRONTEND" != "true" ]; then
+            echo ""
+            echo "=== 前端检查 ==="
+            for item in "${REQUIRED_FRONTEND[@]}"; do
+              name="${item%%:*}"
+              result="${item##*:}"
+              check_job "$name" "$result"
+            done
+          else
+            echo ""
+            echo "=== 前端检查（纯后端PR，跳过）==="
+          fi
+
+          echo ""
+          if [ "$ALL_PASSED" = "true" ]; then
+            echo "✅ CI Gate: PASSED"
+            echo "gate_result=success" >> $GITHUB_OUTPUT
+            exit 0
+          else
+            echo "❌ CI Gate: FAILED"
+            echo "失败项: ${FAILED_ITEMS[*]}"
+            echo "gate_result=failure" >> $GITHUB_OUTPUT
+            exit 1
+          fi
+
+      - name: Report CI trace
+        if: always()
+        shell: sh
+        env:
+          AGENTLOOP_LICENSE_KEY: ${{ secrets.AGENTLOOP_LICENSE_KEY }}
+        run: |
+          STATUS="ok"
+          [ "${{ steps.gate.outputs.gate_result }}" = "success" ] || STATUS="error"
+          START_TIME=""
+          [ -f /tmp/ci_job_start_time ] && START_TIME=$(cat /tmp/ci_job_start_time)
+          python3 scripts/ci/ci_trace_report.py --service xiaoxia-saas-ci --status $STATUS --start-time "$START_TIME" || true
@@ -48,6 +48,7 @@ jobs:
          GITEA_TOKEN: ${{ secrets.REVIEW_GITEA_TOKEN }}
          REPO_NAME: ${{ gitea.repository }}
          PR_NUMBER: ${{ gitea.event.pull_request.number }}
+          PR_HEAD_SHA: ${{ gitea.event.pull_request.head.sha }}
          # LLM 提供商: coze (扣子原生Bot) / openai (OpenAI兼容)
          LLM_PROVIDER: "coze"
          # 扣子模式配置（默认国内站 api.coze.cn）
@@ -60,8 +61,9 @@ jobs:
          LLM_TIMEOUT: "120"
        run: |
          python3 scripts/ci_code_review.py
-        # 审查脚本异常不影响 CI 通过
-        continue-on-error: true
+        # 注意：脚本退出码决定job状态
+        # - 有阻塞级问题 → exit 1 → job失败 → 门禁拦截
+        # - 无阻塞级问题/LLM异常 → exit 0 → 通过（fail-open）

      - name: Report CI trace
        if: always()
@@ -1,4 +1,5 @@
 name: Daily Health Check
+# 注意：使用 curl step_checkout.sh 方式以兼容 docker runner

 on:
  schedule:
@@ -23,50 +24,12 @@ jobs:
        env:
          GITHUB_TOKEN: ${{ github.token }}
        run: |
-          set -eu
-          python3 - <<'PY'
-          import io, os, tarfile, time, urllib.request, urllib.error
-          url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
-          request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
-          last_err = None
-          for attempt in range(5):
-              try:
-                  with urllib.request.urlopen(request, timeout=120) as response:
-                      archive = response.read()
-                  break
-              except urllib.error.HTTPError as e:
-                  last_err = e
-                  if e.code >= 500 and attempt < 4:
-                      wait = 2 ** attempt
-                      print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                      time.sleep(wait)
-                      continue
-                  raise
-              except Exception as e:
-                  last_err = e
-                  if attempt < 4:
-                      wait = 2 ** attempt
-                      print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                      time.sleep(wait)
-                      continue
-                  raise
-          else:
-              raise last_err
-          with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
-              root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
-              for member in tar.getmembers():
-                  name = member.name
-                  if name == root_prefix[:-1]:
-                      continue
-                  if name.startswith(root_prefix):
-                      member.name = name[len(root_prefix):]
-                  if member.name:
-                      tar.extract(member, '.')
-          PY
-
+          curl -sH "Authorization: token $GITHUB_TOKEN" \
+            "${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/raw/scripts/ci/step_checkout.sh?ref=${GITHUB_SHA}" \
+            | bash
      - name: Production health check & smoke test
        id: smoke
-        shell: sh
+        shell: bash
        env:
          SMOKE_ENV: production
          EXISTING_TOKEN: ${{ secrets.PROD_E2E_TOKEN }}
@@ -132,50 +95,12 @@ jobs:
        env:
          GITHUB_TOKEN: ${{ github.token }}
        run: |
-          set -eu
-          python3 - <<'PY'
-          import io, os, tarfile, time, urllib.request, urllib.error
-          url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
-          request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
-          last_err = None
-          for attempt in range(5):
-              try:
-                  with urllib.request.urlopen(request, timeout=120) as response:
-                      archive = response.read()
-                  break
-              except urllib.error.HTTPError as e:
-                  last_err = e
-                  if e.code >= 500 and attempt < 4:
-                      wait = 2 ** attempt
-                      print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                      time.sleep(wait)
-                      continue
-                  raise
-              except Exception as e:
-                  last_err = e
-                  if attempt < 4:
-                      wait = 2 ** attempt
-                      print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                      time.sleep(wait)
-                      continue
-                  raise
-          else:
-              raise last_err
-          with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
-              root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
-              for member in tar.getmembers():
-                  name = member.name
-                  if name == root_prefix[:-1]:
-                      continue
-                  if name.startswith(root_prefix):
-                      member.name = name[len(root_prefix):]
-                  if member.name:
-                      tar.extract(member, '.')
-          PY
-
+          curl -sH "Authorization: token $GITHUB_TOKEN" \
+            "${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/raw/scripts/ci/step_checkout.sh?ref=${GITHUB_SHA}" \
+            | bash
      - name: Run API smoke test on staging
        id: smoke
-        shell: sh
+        shell: bash
        env:
          STAGING_TEST_USER: ${{ secrets.STAGING_TEST_USER }}
          STAGING_TEST_PASSWORD: ${{ secrets.STAGING_TEST_PASSWORD }}
@@ -183,7 +108,8 @@ jobs:
          set +e
          START_TIME=$(date +%s)
          chmod +x tests/e2e/api_smoke_test.sh
-          docker run --rm \
+          CONTAINER_NAME="ci-test-$$"
+          docker create --name "$CONTAINER_NAME" \
            -e BASE_URL=https://staging-api.xiaoxiajianji.com \
            -e WEB_URL=https://staging.xiaoxiajianji.com \
            -e TEST_USER="$STAGING_TEST_USER" \
@@ -192,11 +118,13 @@ jobs:
            -e PERF_CHECK_ENABLED=1 \
            -e PERF_WARN_THRESHOLD_MS=500 \
            -e PERF_FAIL_THRESHOLD_MS=3000 \
-            -v "$PWD:/workspace" \
            -w /workspace \
            git.xiaoxiajianji.com/xiaoxia/base/playwright:v1.45.0-jammy \
-            bash tests/e2e/api_smoke_test.sh 2>&1 | tee /tmp/staging-api-smoke.log
+            bash tests/e2e/api_smoke_test.sh 2>&1
+          docker cp . "$CONTAINER_NAME:/workspace"
+          docker start -a "$CONTAINER_NAME" 2>&1 | tee /tmp/staging-api-smoke.log
          SMOKE_EXIT=${PIPESTATUS[0]}
+          docker rm "$CONTAINER_NAME" > /dev/null 2>&1 || true
          END_TIME=$(date +%s)
          ELAPSED=$((END_TIME - START_TIME))

@@ -218,18 +146,21 @@ jobs:

      - name: Run Staging API Integration Tests (Playwright)
        id: e2e_api
-        shell: sh
+        shell: bash
        run: |
          set +e
          START_TIME=$(date +%s)
-          docker run --rm \
+          CONTAINER_NAME="ci-test-$$"
+          docker create --name "$CONTAINER_NAME" \
            -e E2E_BASE_URL=https://staging.xiaoxiajianji.com \
            -e E2E_API_BASE=https://staging-api.xiaoxiajianji.com/api/v1 \
-            -v "$PWD:/workspace" \
            -w /workspace/apps/web \
            git.xiaoxiajianji.com/xiaoxia/base/playwright:v1.45.0-jammy \
-            sh -lc "npm ci && npx playwright test --reporter=line e2e/test_auth.spec.ts e2e/test_asset.spec.ts e2e/test_project.spec.ts" 2>&1 | tee /tmp/staging-api-e2e.log
+            sh -lc "npm ci && npx playwright test --reporter=line e2e/test_auth.spec.ts e2e/test_asset.spec.ts e2e/test_project.spec.ts" 2>&1
+          docker cp . "$CONTAINER_NAME:/workspace"
+          docker start -a "$CONTAINER_NAME" 2>&1 | tee /tmp/staging-api-e2e.log
          EXIT_CODE=${PIPESTATUS[0]}
+          docker rm "$CONTAINER_NAME" > /dev/null 2>&1 || true
          END_TIME=$(date +%s)
          ELAPSED=$((END_TIME - START_TIME))

@@ -284,63 +215,28 @@ jobs:
        env:
          GITHUB_TOKEN: ${{ github.token }}
        run: |
-          set -eu
-          python3 - <<'PY'
-          import io, os, tarfile, time, urllib.request, urllib.error
-          url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
-          request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
-          last_err = None
-          for attempt in range(5):
-              try:
-                  with urllib.request.urlopen(request, timeout=120) as response:
-                      archive = response.read()
-                  break
-              except urllib.error.HTTPError as e:
-                  last_err = e
-                  if e.code >= 500 and attempt < 4:
-                      wait = 2 ** attempt
-                      print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                      time.sleep(wait)
-                      continue
-                  raise
-              except Exception as e:
-                  last_err = e
-                  if attempt < 4:
-                      wait = 2 ** attempt
-                      print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
-                      time.sleep(wait)
-                      continue
-                  raise
-          else:
-              raise last_err
-          with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
-              root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
-              for member in tar.getmembers():
-                  name = member.name
-                  if name == root_prefix[:-1]:
-                      continue
-                  if name.startswith(root_prefix):
-                      member.name = name[len(root_prefix):]
-                  if member.name:
-                      tar.extract(member, '.')
-          PY
-
+          curl -sH "Authorization: token $GITHUB_TOKEN" \
+            "${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/raw/scripts/ci/step_checkout.sh?ref=${GITHUB_SHA}" \
+            | bash
      - name: Run Playwright E2E on staging
        id: e2e
-        shell: sh
+        shell: bash
        run: |
          set +e
          START_TIME=$(date +%s)
-          docker run --rm --ipc=host \
+          CONTAINER_NAME="ci-test-$$"
+          docker create --name "$CONTAINER_NAME" --ipc=host \
            -e E2E_BASE_URL=https://staging.xiaoxiajianji.com \
            -e E2E_API_BASE=https://staging-api.xiaoxiajianji.com/api/v1 \
            -e E2E_BROWSER_CHANNEL=chromium \
            -e PLAYWRIGHT_HEADLESS=1 \
-            -v "$PWD:/workspace" \
            -w /workspace/apps/web \
            git.xiaoxiajianji.com/xiaoxia/base/playwright:v1.45.0-jammy \
-            sh -lc 'npm ci && npx playwright test --reporter=line --project=chromium e2e/auth.spec.ts e2e/auth-guard.spec.ts e2e/core-upload.spec.ts e2e/core-generation.spec.ts e2e/core-titles.spec.ts' 2>&1 | tee /tmp/staging-e2e.log
+            sh -lc 'npm ci && npx playwright test --reporter=line --project=chromium e2e/auth.spec.ts e2e/auth-guard.spec.ts e2e/core-upload.spec.ts e2e/core-generation.spec.ts e2e/core-titles.spec.ts' 2>&1
+          docker cp . "$CONTAINER_NAME:/workspace"
+          docker start -a "$CONTAINER_NAME" 2>&1 | tee /tmp/staging-e2e.log
          EXIT_CODE=${PIPESTATUS[0]}
+          docker rm "$CONTAINER_NAME" > /dev/null 2>&1 || true
          END_TIME=$(date +%s)
          ELAPSED=$((END_TIME - START_TIME))

@@ -726,4 +622,3 @@ jobs:
          START_TIME=""
          [ -f /tmp/ci_job_start_time ] && START_TIME=$(cat /tmp/ci_job_start_time)
          python3 scripts/ci/ci_trace_report.py --service xiaoxia-saas-ci --status $STATUS --start-time "$START_TIME" || true
-
@@ -1,8 +1,10 @@
 #!/usr/bin/env python3
 """CI中自动修复代码格式（Python: black + isort | Frontend: prettier），并推送回原分支。

- PR事件：自动修复并push回PR源分支（Agent提交的PR自动修，人提交的仅诊断）
+- PR事件：所有PR只要Code Quality因格式问题失败，自动修复并push回源分支
 - Push事件（develop/main）：自动修复并push回原分支，保持主干格式永远正确
+- 防循环：修复commit带 [skip ci-format-check] 标记，检测到该标记则跳过修复
+- 只修格式（black/isort/prettier），ruff逻辑类错误不动
 当code quality检查因格式问题失败时触发。
 """

@@ -239,7 +241,7 @@ def main():
            print("无法获取PR号，跳过自动修复")
            return

-        # 获取PR作者信息，判断是人还是Agent提交的
+        # 获取PR信息
        pr_info_url = f"{api_url}/repos/{repo}/pulls/{pr_number}"
        req_pr = urllib.request.Request(pr_info_url, headers={"Authorization": f"token {token}"})
        with urllib.request.urlopen(req_pr) as resp:
@@ -247,17 +249,26 @@ def main():
        pr_author = pr_info.get("user", {}).get("login", "")
        print(f"PR作者: {pr_author}")

-        # 判断是否为Agent提交的PR
-        agent_authors = {"actions", "auto-approve-bot", "gitea-actions"}
-        is_agent_pr = pr_author in agent_authors or "bot" in pr_author.lower()
+        # 防循环检测：检查最新commit是否已经是格式修复commit
+        # 修复commit message 带 [skip ci-format-check] 标记，检测到则跳过
+        head_branch_tmp = pr_info.get("head", {}).get("ref", "")
+        skip_marker = "[skip ci-format-check]"
+        try:
+            commits_url = f"{api_url}/repos/{repo}/pulls/{pr_number}/commits?limit=3"
+            req_commits = urllib.request.Request(commits_url, headers={"Authorization": f"token {token}"})
+            with urllib.request.urlopen(req_commits) as resp_commits:
+                commits = json.loads(resp_commits.read())
+            latest_msg = commits[0].get("commit", {}).get("message", "") if commits else ""
+            if skip_marker in latest_msg:
+                print(f"检测到最新commit包含 {skip_marker} 标记，跳过格式修复（防循环）")
+                print("本次格式检查失败是格式修复commit触发的CI回跑，属正常现象")
+                sys.exit(0)
+        except Exception as e:
+            print(f"⚠️  防循环检测失败，继续执行: {e}")

-        if is_agent_pr:
-            print(f"检测到Agent提交的PR（作者: {pr_author}），将自动修复并推送")
-            fix_mode = "auto_fix_and_push"
-        else:
-            print(f"检测到人提交的PR（作者: {pr_author}），仅诊断不自动修改")
-            print("（如需自动修复，请用Agent账号提交PR，或手动运行格式化脚本）")
-            fix_mode = "diagnose_only"
+        # 所有PR都自动修复格式（不再区分人/Agent）
+        print("检测到格式问题，将自动修复并推送回分支")
+        fix_mode = "auto_fix_and_push"

        print("=== 检测到代码格式问题，尝试自动修复 ===")
        print(f"PR #{pr_number}")
@@ -315,26 +326,6 @@ def main():
        print("没有需要提交的格式改动")
        return

-    # 诊断模式：只报告问题，不修改不推送
-    if fix_mode == "diagnose_only":
-        print()
-        print("=" * 50)
-        print("📋 格式问题诊断报告（人提交的PR，仅诊断不自动修复）")
-        print("=" * 50)
-        print()
-        print("以下文件存在格式问题，建议手动修复：")
-        for line in result.stdout.strip().split("\n"):
-            print(f"  {line}")
-        print()
-        print("修复方式：")
-        print("  后端(Python): 运行 black + isort")
-        print("  前端: 运行 prettier --write")
-        print("  或使用 scripts/agent-commit.sh 提交（自动格式化）")
-        print()
-        print("=" * 50)
-        # 以非0状态码退出，让CI继续报失败（因为问题没修）
-        sys.exit(1)
-
    print()
    print("变更文件:")
    for line in result.stdout.strip().split("\n"):
@@ -342,7 +333,7 @@ def main():

    # 提交修复
    run("git add -A")
-    run('git commit -m "style: auto-format with black + isort + prettier"')
+    run('git commit -m "style: auto-format with black + isort + prettier [skip ci-format-check]"')

    # 推送（head_branch已从ensure_git_repo获取）
    print(f"\nPR来源分支: {head_branch}")
@@ -23,23 +23,11 @@ FRONTEND_COUNT=$(echo "$FILES" | grep -c '^apps/web/' || true)
 BACKEND_COUNT=$((TOTAL - FRONTEND_COUNT))
 echo "变更文件: ${TOTAL} 个 (前端: ${FRONTEND_COUNT}, 后端/公共: ${BACKEND_COUNT})"

-if [ "$BACKEND_COUNT" = "0" ] && [ "$FRONTEND_COUNT" -gt "0" ]; then
-  CONTEXTS=("CI/CD Pipeline / Frontend Lint (pull_request)")
-  echo "纯前端改动，只检查Frontend Lint"
-else
-  CONTEXTS=(
-    "CI/CD Pipeline / Validate - Code Quality (pull_request)"
-    "CI/CD Pipeline / Validate - Type Check (mypy) (pull_request)"
-    "CI/CD Pipeline / Validate - Migration (alembic) (pull_request)"
-    "CI/CD Pipeline / Frontend Lint (pull_request)"
-    "CI/CD Pipeline / Unit Tests (pull_request)"
-    "CI/CD Pipeline / Frontend Unit Tests (pull_request)"
-    "CI/CD Pipeline / PR Build API Image (pull_request)"
-    "CI/CD Pipeline / PR Build Web Image (pull_request)"
-    "CI/CD Pipeline / PR Build Worker Image (pull_request)"
-  )
-  echo "检查required门禁（与分支保护一致）"
-fi
+# 使用统一的CI Gate门禁（单一检查点，自动处理前端/后端/全栈跳过逻辑）
+CONTEXTS=(
+  "CI/CD Pipeline / CI Gate (pull_request)"
+)
+echo "检查CI Gate统一门禁"
 echo

 # 初始等待30秒，给CI启动写status的时间
@@ -0,0 +1,780 @@
+#!/usr/bin/env python3
+"""
+CI Code Review Script
+- 从 Gitea 获取 PR diff
+- 调用 LLM 进行代码审查
+- 将审查结果写回 PR 评论
+"""
+
+import argparse
+import json
+import logging
+import os
+import re
+import sys
+from typing import Optional, Tuple
+
+import requests
+
+# ============== 日志配置 ==============
+logging.basicConfig(
+    level=logging.INFO,
+    format="[%(asctime)s] [%(levelname)s] %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
+)
+logger = logging.getLogger("ci_code_review")
+
+
+# ============== 常量配置 ==============
+# diff 最大字符数（超过则截断）
+MAX_DIFF_CHARS = int(os.getenv("MAX_DIFF_CHARS", "30000"))
+# LLM 调用超时时间（秒）
+LLM_TIMEOUT = int(os.getenv("LLM_TIMEOUT", "120"))
+# Gitea API 超时时间（秒）
+GITEA_TIMEOUT = int(os.getenv("GITEA_TIMEOUT", "30"))
+# 最大重试次数
+MAX_RETRIES = int(os.getenv("MAX_RETRIES", "2"))
+# LLM 提供商: openai (OpenAI兼容) / coze (扣子原生Bot API)
+LLM_PROVIDER = os.getenv("LLM_PROVIDER", "coze").lower()
+
+
+# ============== 工具函数 ==============
+def truncate_diff(diff_text: str, max_chars: int) -> Tuple[str, bool]:
+    """
+    截断过大的 diff 内容，避免超出 LLM 上下文限制。
+    优先保留文件头和前面的变更，末尾加提示。
+    """
+    if len(diff_text) <= max_chars:
+        return diff_text, False
+
+    # 找到一个合适的截断位置（尽量在文件边界）
+    truncated = diff_text[:max_chars]
+    # 尝试在最后一个 "diff --git" 处截断，避免截断到一半
+    last_file_boundary = truncated.rfind("\ndiff --git ")
+    if last_file_boundary > max_chars // 2:
+        truncated = truncated[:last_file_boundary]
+
+    truncated += (
+        f"\n\n... [DIFF TRUNCATED] 原始 diff 共 {len(diff_text)} 字符，"
+        f"已截断至 {len(truncated)} 字符，仅审查前半部分。\n"
+    )
+    return truncated, True
+
+
+def get_env_or_fail(name: str) -> str:
+    """从环境变量获取值，不存在则报错退出。"""
+    value = os.getenv(name)
+    if not value:
+        logger.error(f"环境变量 {name} 未设置")
+        sys.exit(1)
+    return value
+
+
+# ============== Gitea API 相关 ==============
+class GiteaClient:
+    """Gitea API 客户端"""
+
+    def __init__(self, base_url: str, token: str, repo: str):
+        # 确保 base_url 以 / 结尾
+        self.base_url = base_url.rstrip("/") + "/"
+        self.token = token
+        self.repo = repo  # 格式: owner/repo
+        self.session = requests.Session()
+        self.session.headers.update(
+            {
+                "Authorization": f"token {token}",
+                "Accept": "application/json",
+                "Content-Type": "application/json",
+            }
+        )
+
+    def _api_url(self, path: str) -> str:
+        """拼接 API 路径"""
+        return f"{self.base_url}api/v1/repos/{self.repo}/{path.lstrip('/')}"
+
+    def get_pr_diff(self, pr_number: int) -> str:
+        """
+        获取 PR 的 diff 内容。
+        Gitea API: GET /repos/{owner}/{repo}/pulls/{index}.diff
+        """
+        url = self._api_url(f"pulls/{pr_number}.diff")
+        logger.info(f"获取 PR #{pr_number} diff: {url}")
+
+        resp = self.session.get(
+            url,
+            timeout=GITEA_TIMEOUT,
+            headers={
+                "Accept": "text/plain",
+            },
+        )
+        if resp.status_code != 200:
+            logger.error(f"获取 diff 失败: HTTP {resp.status_code} - {resp.text[:200]}")
+            raise RuntimeError(f"Failed to get PR diff: HTTP {resp.status_code}")
+
+        diff_text = resp.text
+        logger.info(f"获取到 diff，共 {len(diff_text)} 字符")
+        return diff_text
+
+    def get_pr_files(self, pr_number: int) -> list:
+        """
+        获取 PR 修改的文件列表。
+        Gitea API: GET /repos/{owner}/{repo}/pulls/{index}/files
+        """
+        url = self._api_url(f"pulls/{pr_number}/files")
+        logger.info(f"获取 PR #{pr_number} 文件列表")
+
+        resp = self.session.get(url, timeout=GITEA_TIMEOUT)
+        if resp.status_code != 200:
+            logger.warning(f"获取文件列表失败: HTTP {resp.status_code}")
+            return []
+
+        files = resp.json()
+        logger.info(f"PR 修改了 {len(files)} 个文件")
+        return files
+
+    def post_pr_comment(self, pr_number: int, body: str) -> bool:
+        """
+        在 PR 上发布评论。
+        Gitea API: POST /repos/{owner}/{repo}/issues/{index}/comments
+        （Gitea 中 PR 评论走 issues 接口）
+        """
+        url = self._api_url(f"issues/{pr_number}/comments")
+        logger.info(f"发布 PR 评论: {url}")
+
+        payload = {"body": body}
+        resp = self.session.post(
+            url,
+            data=json.dumps(payload),
+            timeout=GITEA_TIMEOUT,
+        )
+        if resp.status_code not in (200, 201):
+            logger.error(f"发布评论失败: HTTP {resp.status_code} - {resp.text[:200]}")
+            return False
+
+        logger.info(f"评论发布成功，评论 ID: {resp.json().get('id', 'unknown')}")
+        return True
+
+    def get_existing_review_comments(self, pr_number: int, marker: str) -> list:
+        """
+        获取 PR 上已有的 AI 审查评论 ID 列表（带标识 marker）。
+        """
+        url = self._api_url(f"issues/{pr_number}/comments")
+        resp = self.session.get(url, timeout=GITEA_TIMEOUT)
+        if resp.status_code != 200:
+            logger.warning(f"获取评论列表失败: HTTP {resp.status_code}")
+            return []
+
+        comments = resp.json()
+        review_comment_ids = []
+        for c in comments:
+            body = c.get("body", "")
+            if marker in body:
+                review_comment_ids.append(c.get("id"))
+        logger.info(f"找到 {len(review_comment_ids)} 条旧的 AI 审查评论")
+        return review_comment_ids
+
+    def delete_pr_comment(self, pr_number: int, comment_id: int) -> bool:
+        """
+        删除 PR 上的指定评论。
+        """
+        url = self._api_url(f"issues/comments/{comment_id}")
+        resp = self.session.delete(url, timeout=GITEA_TIMEOUT)
+        if resp.status_code not in (200, 204):
+            logger.warning(f"删除评论 {comment_id} 失败: HTTP {resp.status_code}")
+            return False
+        return True
+
+    def create_commit_status(
+        self, sha: str, state: str, context: str, description: str = "", target_url: str = ""
+    ) -> bool:
+        """
+        给指定 commit 打 status。
+        state: pending / success / failure / error / warning
+        Gitea API: POST /repos/{owner}/{repo}/statuses/{sha}
+        """
+        url = self._api_url(f"statuses/{sha}")
+        logger.info(f"设置 commit status: sha={sha[:12]}..., state={state}, context={context}")
+
+        payload = {
+            "state": state,
+            "context": context,
+            "description": description[:200] if description else "",
+        }
+        if target_url:
+            payload["target_url"] = target_url
+
+        resp = self.session.post(
+            url,
+            data=json.dumps(payload),
+            timeout=GITEA_TIMEOUT,
+        )
+        if resp.status_code not in (200, 201):
+            logger.error(f"设置 status 失败: HTTP {resp.status_code} - {resp.text[:200]}")
+            return False
+
+        logger.info(f"Status 设置成功: {context} = {state}")
+        return True
+
+
+def call_llm_openai(
+    prompt: str,
+    llm_base_url: str,
+    llm_api_key: str,
+    llm_model: str,
+) -> Optional[str]:
+    """OpenAI 兼容模式调用"""
+    base_url = llm_base_url.rstrip("/") + "/"
+    api_url = f"{base_url}chat/completions"
+
+    headers = {
+        "Authorization": f"Bearer {llm_api_key}",
+        "Content-Type": "application/json",
+    }
+
+    payload = {
+        "model": llm_model,
+        "messages": [
+            {
+                "role": "system",
+                "content": "你是一位严谨的资深代码审查专家，擅长发现代码中的逻辑错误、安全隐患和性能问题。",
+            },
+            {
+                "role": "user",
+                "content": prompt,
+            },
+        ],
+        "temperature": 0.3,
+        "max_tokens": 2048,
+    }
+
+    logger.info(f"调用 LLM (OpenAI兼容): {api_url}, model={llm_model}")
+
+    last_error = None
+    for attempt in range(MAX_RETRIES + 1):
+        try:
+            resp = requests.post(
+                api_url,
+                headers=headers,
+                json=payload,
+                timeout=LLM_TIMEOUT,
+            )
+            if resp.status_code != 200:
+                logger.warning(f"LLM 调用失败 (第 {attempt + 1} 次): " f"HTTP {resp.status_code} - {resp.text[:200]}")
+                last_error = f"HTTP {resp.status_code}"
+                continue
+
+            data = resp.json()
+            choices = data.get("choices", [])
+            if not choices:
+                logger.warning(f"LLM 返回空结果 (第 {attempt + 1} 次)")
+                last_error = "empty choices"
+                continue
+
+            content = choices[0].get("message", {}).get("content", "")
+            if not content.strip():
+                logger.warning(f"LLM 返回空内容 (第 {attempt + 1} 次)")
+                last_error = "empty content"
+                continue
+
+            logger.info(f"LLM 审查完成，结果长度: {len(content)} 字符")
+            return content
+
+        except requests.Timeout:
+            logger.warning(f"LLM 调用超时 (第 {attempt + 1} 次)")
+            last_error = "timeout"
+        except requests.RequestException as e:
+            logger.warning(f"LLM 调用异常 (第 {attempt + 1} 次): {e}")
+            last_error = str(e)
+
+    logger.error(f"LLM 调用最终失败: {last_error}")
+    return None
+
+
+def call_llm_coze(
+    prompt: str,
+    llm_base_url: str,
+    llm_api_key: str,
+    llm_model: str,
+    coze_bot_id: str,
+) -> Optional[str]:
+    """扣子(Coze)原生 Bot API 调用（支持异步轮询）"""
+    import time
+
+    base_url = llm_base_url.rstrip("/") + "/"
+    api_url = f"{base_url}v3/chat"
+
+    headers = {
+        "Authorization": f"Bearer {llm_api_key}",
+        "Content-Type": "application/json",
+    }
+
+    payload = {
+        "bot_id": coze_bot_id,
+        "user_id": "ci-code-review-bot",
+        "stream": False,
+        "additional_messages": [
+            {
+                "role": "user",
+                "content": prompt,
+                "content_type": "text",
+            }
+        ],
+    }
+
+    logger.info(f"调用 LLM (Coze): {api_url}, bot_id={coze_bot_id}")
+
+    last_error = None
+    for attempt in range(MAX_RETRIES + 1):
+        try:
+            resp = requests.post(
+                api_url,
+                headers=headers,
+                json=payload,
+                timeout=LLM_TIMEOUT,
+            )
+            if resp.status_code != 200:
+                logger.warning(f"Coze 调用失败 (第 {attempt + 1} 次): " f"HTTP {resp.status_code} - {resp.text[:300]}")
+                last_error = f"HTTP {resp.status_code}"
+                continue
+
+            data = resp.json()
+            chat_data = data.get("data", {})
+            chat_id = chat_data.get("id", "")
+            conversation_id = chat_data.get("conversation_id", "")
+            status = chat_data.get("status", "")
+
+            # Coze v3 API 异步：先返回 in_progress，需要轮询
+            if status == "in_progress" and conversation_id and chat_id:
+                logger.info(f"Coze 异步处理中，开始轮询... (chat_id={chat_id[:12]}...)")
+                # 轮询 message 列表接口（GET + query参数），最多等 LLM_TIMEOUT 秒
+                poll_url = f"{base_url}v3/chat/message/list"
+                poll_start = time.time()
+                poll_interval = 3  # 每3秒轮询一次
+
+                while time.time() - poll_start < LLM_TIMEOUT:
+                    time.sleep(poll_interval)
+                    poll_params = {
+                        "chat_id": chat_id,
+                        "conversation_id": conversation_id,
+                    }
+                    poll_resp = requests.get(
+                        poll_url,
+                        headers=headers,
+                        params=poll_params,
+                        timeout=GITEA_TIMEOUT,
+                    )
+                    if poll_resp.status_code != 200:
+                        logger.debug(f"轮询返回 HTTP {poll_resp.status_code}: {poll_resp.text[:100]}")
+                        continue
+
+                    poll_data = poll_resp.json()
+                    if poll_data.get("code", 0) != 0:
+                        logger.debug(f"轮询返回错误: {poll_data.get('msg', '')}")
+                        continue
+
+                    messages = poll_data.get("data", []) or []
+
+                    # 找assistant的answer消息
+                    content = None
+                    for msg in messages:
+                        if msg.get("role") == "assistant" and msg.get("type") == "answer":
+                            content = msg.get("content", "")
+                            break
+
+                    if content and content.strip():
+                        logger.info(f"Coze 审查完成，结果长度: {len(content)} 字符")
+                        return content
+
+                logger.warning(f"Coze 轮询超时 ({LLM_TIMEOUT}s)，未拿到结果")
+                last_error = "poll timeout"
+                continue
+
+            # 同步返回的情况（兼容）
+            content = None
+            messages = chat_data.get("messages", []) or data.get("messages", [])
+            for msg in messages:
+                if msg.get("role") == "assistant" and msg.get("type") == "answer":
+                    content = msg.get("content", "")
+                    break
+
+            if not content:
+                content = chat_data.get("content") or data.get("content")
+
+            if not content:
+                choices = data.get("choices", [])
+                if choices:
+                    content = choices[0].get("message", {}).get("content", "")
+
+            if not content or not content.strip():
+                logger.warning(f"Coze 返回空内容 (第 {attempt + 1} 次): {str(data)[:200]}")
+                last_error = "empty content"
+                continue
+
+            logger.info(f"Coze 审查完成，结果长度: {len(content)} 字符")
+            return content
+
+        except requests.Timeout:
+            logger.warning(f"Coze 调用超时 (第 {attempt + 1} 次)")
+            last_error = "timeout"
+        except requests.RequestException as e:
+            logger.warning(f"Coze 调用异常 (第 {attempt + 1} 次): {e}")
+            last_error = str(e)
+
+    logger.error(f"Coze 调用最终失败: {last_error}")
+    return None
+
+
+def build_review_prompt(diff_text: str, pr_number: int, file_list: list) -> str:
+    """
+    构建代码审查的 Prompt。
+    包含：PR 基本信息、修改文件列表、diff 内容、审查要求。
+    """
+    # 提取文件名列表
+    file_names = [f.get("filename", "") for f in file_list] if file_list else []
+    file_list_str = "\n".join(f"  - {fn}" for fn in file_names) if file_names else "  （未获取到文件列表）"
+
+    prompt = f"""请作为资深代码审查专家，对以下 Pull Request 的代码变更进行严格审查。
+
+## PR 基本信息
+- PR 编号: #{pr_number}
+- 修改文件数: {len(file_list) if file_list else '未知'}
+
+## 修改文件列表
+{file_list_str}
+
+## 代码变更（diff）
+```diff
+{diff_text}
+```
+
+## 审查要求
+请从以下维度进行审查，重点关注**阻塞级问题**：
+
+### 问题分级标准
+- **🔴 阻塞级（BLOCKER）**：必须修复，否则不允许合并。包括：
+  1. **明显逻辑bug**：条件判断错误、死循环、返回值错误、空指针/None引用未处理、边界条件遗漏导致功能异常
+  2. **安全漏洞**：SQL注入、XSS、命令注入、敏感信息明文存储/泄露、权限绕过、认证缺失
+  3. **语法错误**：代码存在语法层面的错误，无法运行
+  4. **数据损坏风险**：可能导致数据丢失、数据不一致、脏数据写入的问题
+
+- **💡 建议级（SUGGESTION）**：不阻塞合并，仅供参考改进。包括：
+  1. 命名不规范、代码风格问题
+  2. 最佳实践建议、设计模式优化
+  3. 格式问题（缩进、空行、import顺序等）
+  4. 代码可读性改进、注释补充
+  5. 非关键路径的轻微性能优化建议
+  6. 重复代码、过长函数等代码质量问题
+
+1. **逻辑正确性**：是否有明显的逻辑错误、边界条件遗漏、空指针/None引用风险
+2. **异常处理**：异常捕获是否合理，是否有裸except，错误处理是否完善
+3. **参数校验**：函数入参、返回值是否有必要的校验
+4. **代码质量**：是否有重复代码、命名不清晰、过于复杂的函数
+5. **性能问题**：是否有明显的性能隐患（如循环内重复计算、不必要的数据库查询）
+6. **安全问题**：是否有注入风险、敏感信息泄露、权限控制问题
+
+## 输出格式
+请使用以下格式输出，语言为中文。**必须严格按照格式输出，尤其是【阻塞级判定】部分**：
+
+### 【阻塞级判定】
+- 是否存在阻塞级问题：（是 / 否）
+- 阻塞级问题数量：X 个
+
+### 📊 审查概览
+- 整体评价：（通过 / 有建议 / 需修改）
+- 建议级问题数量：X 个
+
+### 🔴 阻塞级问题（必须修复）
+（如果没有阻塞级问题，写"无"）
+1. **[文件: 行号] 问题标题**
+   - 问题类型：（逻辑bug / 安全漏洞 / 语法错误 / 数据损坏风险）
+   - 问题描述：...
+   - 修改建议：...
+
+### 💡 改进建议（不阻塞合并）
+（如果没有建议，写"无"）
+1. **[文件: 行号] 建议标题**
+   - 具体内容：...
+
+### ✅ 良好实践
+（可选，列出值得肯定的地方）
+
+请务必基于代码实际内容审查，不要编造不存在的问题。如果代码质量良好，直接给出通过结论即可。
+**重要：【阻塞级判定】必须准确，只有确实存在严重问题时才写"是"。**
+"""
+    return prompt
+
+
+def parse_blocker_result(review_text: str) -> Tuple[bool, int]:
+    """
+    从审查结果中解析是否存在阻塞级问题。
+    返回 (has_blocker, blocker_count)
+    """
+    # 先找【阻塞级判定】部分的明确标记
+    pattern = r"【阻塞级判定】[\s\S]*?是否存在阻塞级问题[：:]\s*(是|否)"
+    match = re.search(pattern, review_text)
+    if match:
+        has_blocker = match.group(1) == "是"
+    else:
+        # fallback 1: 找"阻塞级问题数量"
+        count_pattern = r"阻塞级问题数量[：:]\s*(\d+)"
+        count_match = re.search(count_pattern, review_text)
+        if count_match:
+            has_blocker = int(count_match.group(1)) > 0
+        else:
+            # fallback 2: 检查是否有"阻塞级问题"section且内容不是"无"
+            has_blocker = False
+            blocker_section = re.search(r"### 🔴 阻塞级问题[\s\S]*?(?=### |\Z)", review_text)
+            if blocker_section:
+                section_text = blocker_section.group(0)
+                # 如果有编号列表项，说明有问题
+                if re.search(r"\d+\.\s*\*\*", section_text):
+                    has_blocker = True
+
+    # 提取数量
+    count_pattern = r"阻塞级问题数量[：:]\s*(\d+)"
+    count_match = re.search(count_pattern, review_text)
+    blocker_count = int(count_match.group(1)) if count_match else (1 if has_blocker else 0)
+
+    logger.info(f"阻塞级问题解析: 存在={has_blocker}, 数量={blocker_count}")
+    return has_blocker, blocker_count
+
+
+def call_llm_for_review(
+    diff_text: str,
+    pr_number: int,
+    file_list: list,
+    llm_base_url: str,
+    llm_api_key: str,
+    llm_model: str,
+    coze_bot_id: str = "",
+) -> Optional[str]:
+    """
+    调用 LLM 进行代码审查，返回审查结果文本。
+    失败时返回 None。
+    根据 LLM_PROVIDER 环境变量选择调用方式。
+    """
+    prompt = build_review_prompt(diff_text, pr_number, file_list)
+    logger.info(f"Prompt 长度: {len(prompt)} 字符")
+
+    provider = LLM_PROVIDER
+
+    if provider == "coze":
+        return call_llm_coze(prompt, llm_base_url, llm_api_key, llm_model, coze_bot_id)
+    else:
+        # 默认 OpenAI 兼容
+        return call_llm_openai(prompt, llm_base_url, llm_api_key, llm_model)
+
+
+# ============== 主流程 ==============
+def main():
+    parser = argparse.ArgumentParser(description="CI AI 代码审查脚本")
+    parser.add_argument("--pr", type=int, help="PR 编号（也可通过 PR_NUMBER 环境变量）")
+    parser.add_argument("--repo", type=str, help="仓库名 owner/repo（也可通过 REPO_NAME 环境变量）")
+    parser.add_argument("--gitea-url", type=str, help="Gitea 地址（也可通过 GITEA_API_URL 环境变量）")
+    parser.add_argument("--gitea-token", type=str, help="Gitea Token（也可通过 GITEA_TOKEN 环境变量）")
+    parser.add_argument("--dry-run", action="store_true", help="只输出审查结果，不发表评论")
+    args = parser.parse_args()
+
+    # 读取配置
+    gitea_url = args.gitea_url or os.getenv("GITEA_API_URL") or os.getenv("GITEA_SERVER_URL")
+    gitea_token = args.gitea_token or os.getenv("GITEA_TOKEN")
+    repo_name = args.repo or os.getenv("REPO_NAME") or os.getenv("GITEA_REPO")
+    pr_number = args.pr or int(os.getenv("PR_NUMBER") or os.getenv("GITEA_PR_NUMBER") or 0)
+
+    llm_base_url = os.getenv("LLM_BASE_URL")
+    llm_api_key = os.getenv("LLM_API_KEY")
+    llm_model = os.getenv("LLM_MODEL", "")
+    coze_bot_id = os.getenv("COZE_BOT_ID", os.getenv("COZE_BOTID", ""))
+
+    # 根据 provider 设置默认值
+    provider = LLM_PROVIDER
+    if provider == "coze":
+        # 扣子模式：默认国内站，key 兼容多种环境变量名
+        if not llm_base_url:
+            llm_base_url = "https://api.coze.cn"
+        if not llm_api_key:
+            llm_api_key = os.getenv("COZE_API_KEY", "") or os.getenv("COZE_PAT", "")
+    else:
+        # OpenAI兼容模式：默认模型
+        if not llm_model:
+            llm_model = "gpt-4o-mini"
+
+    # 必要参数校验
+    missing = []
+    if not gitea_url:
+        missing.append("GITEA_API_URL")
+    if not gitea_token:
+        missing.append("GITEA_TOKEN")
+    if not repo_name:
+        missing.append("REPO_NAME")
+    if not pr_number:
+        missing.append("PR_NUMBER")
+    if not llm_base_url:
+        missing.append("LLM_BASE_URL")
+    if not llm_api_key:
+        missing.append("LLM_API_KEY")
+    if provider == "coze" and not coze_bot_id:
+        missing.append("COZE_BOT_ID (扣子模式需要)")
+
+    if missing:
+        logger.error(f"缺少必要配置: {', '.join(missing)}")
+        sys.exit(1)
+
+    logger.info(f"开始审查 PR #{pr_number}，仓库: {repo_name}")
+    logger.info(f"Gitea: {gitea_url}")
+    logger.info(f"LLM: {llm_base_url} (model={llm_model})")
+
+    try:
+        # 1. 初始化 Gitea 客户端
+        gitea = GiteaClient(gitea_url, gitea_token, repo_name)
+
+        # 2. 获取 PR diff 和文件列表
+        try:
+            diff_text = gitea.get_pr_diff(pr_number)
+            file_list = gitea.get_pr_files(pr_number)
+        except Exception as e:
+            logger.error(f"获取 PR 信息失败: {e}")
+            sys.exit(1)
+
+        # 3. 过滤掉不需要审查的文件（如 lock 文件、生成的文件、二进制文件等）
+        skip_extensions = (
+            ".lock",
+            ".sum",
+            ".min.js",
+            ".min.css",
+            ".map",
+            ".png",
+            ".jpg",
+            ".jpeg",
+            ".gif",
+            ".svg",
+            ".ico",
+            ".woff",
+            ".woff2",
+            ".ttf",
+            ".eot",
+        )
+        skipped_files = []
+        if file_list:
+            skipped_files = [
+                f.get("filename")
+                for f in file_list
+                if f.get("filename", "").endswith(skip_extensions) or f.get("status") == "removed"
+            ]
+            if skipped_files:
+                logger.info(f"跳过 {len(skipped_files)} 个非文本/已删除文件: {', '.join(skipped_files[:5])}...")
+
+        # 实际从 diff 中移除跳过的文件（按文件边界切割）
+        if skipped_files:
+            diff_lines = diff_text.split("\n")
+            filtered_lines = []
+            current_file = None
+            skip_current = False
+            i = 0
+            while i < len(diff_lines):
+                line = diff_lines[i]
+                # 检测新文件开始: diff --git a/xxx b/xxx
+                if line.startswith("diff --git "):
+                    # 提取文件名
+                    parts = line.split(" ")
+                    if len(parts) >= 4:
+                        # b/ 后面的是目标文件名
+                        current_file = parts[3][2:] if parts[3].startswith("b/") else parts[3]
+                        skip_current = any(current_file == sf for sf in skipped_files) or any(
+                            current_file.endswith(ext) for ext in skip_extensions
+                        )
+                    else:
+                        skip_current = False
+                if not skip_current:
+                    filtered_lines.append(line)
+                i += 1
+            original_len = len(diff_text)
+            diff_text = "\n".join(filtered_lines)
+            logger.info(f"Diff 过滤后: {original_len} -> {len(diff_text)} 字符 (减少 {original_len - len(diff_text)})")
+
+        # 4. 截断过大的 diff
+        diff_text, was_truncated = truncate_diff(diff_text, MAX_DIFF_CHARS)
+        if was_truncated:
+            logger.warning(f"Diff 过大，已截断至 {len(diff_text)} 字符")
+
+        # 5. 如果 diff 为空，直接跳过
+        if not diff_text.strip():
+            logger.info("Diff 为空，无需审查")
+            sys.exit(0)
+
+        # 6. 调用 LLM 审查
+        review_result = call_llm_for_review(
+            diff_text=diff_text,
+            pr_number=pr_number,
+            file_list=file_list,
+            llm_base_url=llm_base_url,
+            llm_api_key=llm_api_key,
+            llm_model=llm_model,
+            coze_bot_id=coze_bot_id,
+        )
+
+        if not review_result:
+            logger.error("LLM 审查失败")
+            sys.exit(0)  # fail-open: LLM调用失败不阻塞合并
+
+        # 7. 加上审查时间和标识（便于识别是自动审查）
+        from datetime import datetime
+
+        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        marker = "<!-- AI_CODE_REVIEW_AUTO_COMMENT -->"
+        full_comment = f"""{review_result}
+
+---
+<sub>🤖 由 AI 代码审查机器人自动生成 | {timestamp} | 模型: {llm_model}</sub>
+
+{marker}
+"""
+
+        # 8. 输出审查结果到日志
+        logger.info("=" * 60)
+        logger.info("审查结果:")
+        for line in review_result.split("\n")[:30]:
+            logger.info(line)
+        if len(review_result.split("\n")) > 30:
+            logger.info(f"... 共 {len(review_result.split(chr(10)))} 行")
+        logger.info("=" * 60)
+
+        # 9. 发布评论（先删除旧的审查评论，避免刷屏）
+        if args.dry_run:
+            logger.info("--dry-run 模式，跳过发布评论")
+            print(full_comment)
+        else:
+            # 去重：删除之前的 AI 审查评论
+            old_comments = gitea.get_existing_review_comments(pr_number, marker)
+            if old_comments:
+                logger.info(f"找到 {len(old_comments)} 条旧的 AI 审查评论，先删除")
+                for cid in old_comments:
+                    gitea.delete_pr_comment(pr_number, cid)
+            # 发布新评论
+            success = gitea.post_pr_comment(pr_number, full_comment)
+            if not success:
+                logger.error("评论发布失败")
+                sys.exit(1)
+
+        # 10. 解析阻塞级问题，用退出码决定 job 状态
+        # 有阻塞级问题 → exit 1 → job失败 → Gitea自动打failure status → 门禁拦截
+        # 无阻塞级问题 → exit 0 → job成功 → Gitea自动打success status
+        # LLM调用失败等异常 → exit 0 → fail-open，不阻塞正常开发
+        has_blocker, blocker_count = parse_blocker_result(review_result)
+
+        if has_blocker:
+            logger.error(f"检测到 {blocker_count} 个阻塞级问题，审查不通过")
+            logger.info("代码审查完成（失败）")
+            sys.exit(1)
+        else:
+            logger.info("无阻塞级问题，审查通过")
+            logger.info("代码审查完成（通过）")
+            sys.exit(0)
+
+    except Exception as e:
+        logger.exception(f"审查脚本发生未预期的异常: {e}")
+        sys.exit(0)  # fail-open: 异常不阻塞正常开发
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,239 @@
+#!/usr/bin/env python3
+"""
+CI触发可靠性监控 - 定时检查PR的CI触发状态
+- 监控open PR的最新commit是否在5分钟内触发了CI
+- 异常时通过飞书webhook告警
+
+环境变量:
+  GITEA_API_TOKEN - Gitea API Token (必填)
+  GITEA_REPO - 仓库路径，如 xiaoxia/xiaoxia-saas
+  GITEA_URL - Gitea地址，如 https://git.xiaoxiajianji.com
+  CI_NOTIFY_WEBHOOK - 飞书告警webhook (必填)
+  CHECK_INTERVAL_MIN - 检查间隔（分钟），默认5
+  STALE_THRESHOLD_MIN - CI未触发告警阈值（分钟），默认5
+"""
+
+import json
+import os
+import sys
+import time
+import urllib.error
+import urllib.request
+
+
+def get_env(name, default=""):
+    return os.environ.get(name, default)
+
+
+def api_get(path):
+    """调用Gitea API"""
+    token = get_env("GITEA_API_TOKEN")
+    base_url = get_env("GITEA_URL", "https://git.xiaoxiajianji.com")
+    repo = get_env("GITEA_REPO", "xiaoxia/xiaoxia-saas")
+
+    url = f"{base_url}/api/v1/repos/{repo}{path}"
+    req = urllib.request.Request(url)
+    req.add_header("Authorization", f"token {token}")
+
+    for attempt in range(3):
+        try:
+            with urllib.request.urlopen(req, timeout=30) as resp:
+                return json.loads(resp.read())
+        except urllib.error.HTTPError as e:
+            if e.code >= 500 and attempt < 2:
+                time.sleep(2**attempt)
+                continue
+            raise
+        except Exception:
+            if attempt < 2:
+                time.sleep(2**attempt)
+                continue
+            raise
+
+
+def get_open_prs():
+    """获取所有open PR"""
+    prs = []
+    page = 1
+    while True:
+        batch = api_get(f"/pulls?state=open&sort=updated&direction=desc&limit=50&page={page}")
+        if not batch:
+            break
+        prs.extend(batch)
+        if len(batch) < 50:
+            break
+        page += 1
+    return prs
+
+
+def get_commit_status(sha):
+    """获取commit的CI状态"""
+    try:
+        return api_get(f"/commits/{sha}/status")
+    except Exception as e:
+        print(f"  ⚠️  获取commit状态失败: {e}")
+        return {"state": "error", "statuses": []}
+
+
+def has_ci_started(statuses):
+    """判断是否有CI job已经启动（pending/running/success/failure都算启动了）"""
+    pr_statuses = [s for s in statuses if "pull_request" in s.get("context", "")]
+    if not pr_statuses:
+        return False
+    # 只要有非pending且非空的状态，就算启动了
+    for s in pr_statuses:
+        if s.get("status") in ["success", "failure", "running"]:
+            return True
+        if s.get("status") == "pending" and "Has started running" in s.get("description", ""):
+            return True
+    # 全是"Blocked by required conditions"的pending也算（说明CI系统收到了事件）
+    for s in pr_statuses:
+        if "Blocked" in s.get("description", ""):
+            return True
+    return False
+
+
+def send_alert(pr_num, pr_title, pr_url, head_sha, commit_age_min):
+    """发送飞书告警"""
+    webhook = get_env("CI_NOTIFY_WEBHOOK")
+    if not webhook:
+        print("  ⚠️  未配置CI_NOTIFY_WEBHOOK，跳过告警")
+        return
+
+    get_env("GITEA_URL", "https://git.xiaoxiajianji.com")
+
+    content = {
+        "msg_type": "interactive",
+        "card": {
+            "header": {
+                "title": {"tag": "plain_text", "content": f"⚠️ CI告警 - PR#{pr_num} CI未触发"},
+                "template": "red",
+            },
+            "elements": [
+                {
+                    "tag": "div",
+                    "text": {
+                        "tag": "lark_md",
+                        "content": f"**PR**: [{pr_title}]({pr_url})\n**最新commit**: `{head_sha[:12]}`\n**已等待**: {commit_age_min:.0f} 分钟仍无CI启动\n**可能原因**: Gitea Actions事件丢失 / Webhook失败 / Runner资源不足",
+                    },
+                },
+                {
+                    "tag": "action",
+                    "actions": [
+                        {
+                            "tag": "button",
+                            "text": {"tag": "plain_text", "content": "查看PR"},
+                            "url": pr_url,
+                            "type": "primary",
+                        },
+                        {
+                            "tag": "button",
+                            "text": {"tag": "plain_text", "content": "查看Actions"},
+                            "url": f"{pr_url}/files",
+                            "type": "default",
+                        },
+                    ],
+                },
+                {
+                    "tag": "note",
+                    "elements": [
+                        {"tag": "plain_text", "content": f"CI触发监控 | 检测时间: {time.strftime('%Y-%m-%d %H:%M:%S')}"}
+                    ],
+                },
+            ],
+        },
+    }
+
+    try:
+        data = json.dumps(content).encode()
+        req = urllib.request.Request(webhook, data=data, method="POST")
+        req.add_header("Content-Type", "application/json")
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            resp.read()
+        print(f"  📢 告警已发送: PR#{pr_num}")
+    except Exception as e:
+        print(f"  ⚠️  告警发送失败: {e}")
+
+
+def main():
+    stale_threshold = int(get_env("STALE_THRESHOLD_MIN", "5"))
+
+    print("=" * 60)
+    print(f"CI触发监控 - 检测时间: {time.strftime('%Y-%m-%d %H:%M:%S')}")
+    print(f"告警阈值: {stale_threshold}分钟无CI启动")
+    print("=" * 60)
+
+    # 获取open PR列表
+    try:
+        prs = get_open_prs()
+    except Exception as e:
+        print(f"❌ 获取PR列表失败: {e}")
+        sys.exit(0)  # 告警脚本不阻断CI
+
+    print(f"\n共 {len(prs)} 个open PR\n")
+
+    stale_prs = []
+    now = time.time()
+
+    for pr in prs:
+        pr_num = pr["number"]
+        pr_title = pr["title"]
+        pr_url = pr["html_url"]
+        head_sha = pr["head"]["sha"]
+        updated_at = pr["updated_at"]
+
+        # 解析updated_at（ISO格式）
+        try:
+            # 2026-07-17T09:22:43+08:00
+            from datetime import datetime
+
+            # 简化处理：直接用字符串解析
+            ts_str = updated_at.replace("Z", "+00:00")
+            # 手动解析
+            dt = datetime.fromisoformat(ts_str)
+            commit_time = dt.timestamp()
+        except Exception as e:
+            print(f"  ⚠️  PR#{pr_num} 时间解析失败: {e}")
+            continue
+
+        age_min = (now - commit_time) / 60
+
+        print(f"PR#{pr_num:3d} | {pr_title[:45]:45s} | 更新于 {age_min:.0f}min前")
+
+        # 少于2分钟的跳过，给CI一点启动时间
+        if age_min < 2:
+            print("  ⏳ 刚更新，等待CI启动...")
+            continue
+
+        # 获取commit状态
+        status = get_commit_status(head_sha)
+        statuses = status.get("statuses", [])
+
+        if has_ci_started(statuses):
+            print(f"  ✅ CI已启动 (state={status.get('state')})")
+            continue
+
+        # CI未启动，判断是否超过阈值
+        if age_min >= stale_threshold:
+            print(f"  🚨 CI未触发！已等待 {age_min:.0f} 分钟")
+            stale_prs.append({"num": pr_num, "title": pr_title, "url": pr_url, "sha": head_sha, "age_min": age_min})
+        else:
+            print(f"  ⏳ CI尚未启动 ({age_min:.0f}min < {stale_threshold}min阈值)")
+
+    # 发送告警
+    print(f"\n{'=' * 60}")
+    print(f"检测结果: {len(stale_prs)} 个PR CI未触发超过阈值")
+
+    if stale_prs:
+        print("\n告警列表:")
+        for pr in stale_prs:
+            print(f"  - PR#{pr['num']}: {pr['title'][:40]} ({pr['age_min']:.0f}min)")
+            send_alert(pr["num"], pr["title"], pr["url"], pr["sha"], pr["age_min"])
+    else:
+        print("✅ 所有PR CI触发正常")
+
+    print("=" * 60)
+
+
+if __name__ == "__main__":
+    main()