fix(queue-limit): P1修复 - 阈值抽常量 + 边界统一 + 入队后兜底校验

- 阈值统一管理：USER_PENDING_LIMIT/GLOBAL_PENDING_LIMIT 抽到 task_enqueue.py 常量，所有入口引用 - 边界判断统一：预检查用 >=（任务创建前），入队检查用 >（包含当前任务），语义一致 - 并发竞态兜底：发送Celery后再查一次DB计数，超限则回滚任务为failed - 新增 12 个单元测试（入队后兜底5个 + 边界验证7个）
feat: 任务队列限流防护 - 用户级3个/全局20个
2026-07-11 16:24:58 +08:00 · 2026-07-11 16:13:11 +08:00
8 changed files with 504 additions and 346 deletions
@@ -1,77 +0,0 @@
-#!/bin/sh
-# 飞书通知脚本 — CI 流水线调用
-# 用法: send_feishu_notify.sh <success|failure> <job_name>
-# 依赖环境变量: CI_NOTIFY_WEBHOOK, GITHUB_REF_NAME, GITHUB_SHA, GITHUB_ACTOR, GITHUB_RUN_ID, GITHUB_REPOSITORY
-
-set -eu
-
-STATUS="$1"
-JOB_NAME="$2"
-WEBHOOK="${CI_NOTIFY_WEBHOOK:-}"
-
-if [ -z "$WEBHOOK" ]; then
-  echo "未配置 CI_NOTIFY_WEBHOOK，跳过通知"
-  exit 0
-fi
-
-BRANCH="${GITHUB_REF_NAME:-unknown}"
-COMMIT="${GITHUB_SHA:-unknown}"
-COMMIT_SHORT="${COMMIT:0:8}"
-ACTOR="${GITHUB_ACTOR:-unknown}"
-RUN_ID="${GITHUB_RUN_ID:-unknown}"
-REPO="${GITHUB_REPOSITORY:-unknown}"
-RUN_URL="https://git.xiaoxiajianji.com/${REPO}/actions/runs/${RUN_ID}"
-
-if [ "$STATUS" = "success" ]; then
-  TITLE="✅ CI告警：${JOB_NAME} 成功"
-  TEMPLATE="green"
-  BUTTON_TEXT="查看详情"
-  BUTTON_TYPE="primary"
-  NOTE_TEXT="流水线执行成功"
-else
-  TITLE="⚠️ CI告警：${JOB_NAME} 失败"
-  TEMPLATE="red"
-  BUTTON_TEXT="查看失败日志"
-  BUTTON_TYPE="danger"
-  NOTE_TEXT="请提交者尽快查看修复！"
-fi
-
-# 用 printf 拼接 JSON，避免 heredoc 缩进问题
-PAYLOAD=$(printf '{
-  "msg_type": "interactive",
-  "card": {
-    "header": {
-      "title": { "tag": "plain_text", "content": "%s" },
-      "template": "%s"
-    },
-    "elements": [
-      {
-        "tag": "div",
-        "text": {
-          "tag": "lark_md",
-          "content": "**项目**: xiaoxia-saas\\n**任务**: %s\\n**分支**: %s\\n**提交**: %s\\n**提交者**: %s\\n**Run**: #%s\\n**[查看日志](%s)**"
-        }
-      },
-      {
-        "tag": "action",
-        "actions": [
-          {
-            "tag": "button",
-            "text": { "tag": "plain_text", "content": "%s" },
-            "url": "%s",
-            "type": "%s"
-          }
-        ]
-      },
-      {
-        "tag": "note",
-        "elements": [
-          { "tag": "plain_text", "content": "%s" }
-        ]
-      }
-    ]
-  }
-}' "$TITLE" "$TEMPLATE" "$JOB_NAME" "$BRANCH" "$COMMIT_SHORT" "$ACTOR" "$RUN_ID" "$RUN_URL" "$BUTTON_TEXT" "$RUN_URL" "$BUTTON_TYPE" "$NOTE_TEXT")
-
-RESPONSE=$(curl -s -X POST -H "Content-Type: application/json" "$WEBHOOK" -d "$PAYLOAD")
-echo "通知响应: $RESPONSE"
@@ -0,0 +1,65 @@
+name: Auto Merge PRs
+
+on:
+  schedule:
+    - cron: '0 */6 * * *'
+  workflow_dispatch:
+
+jobs:
+  auto-merge:
+    runs-on: saas
+    timeout-minutes: 10
+    steps:
+      - name: Checkout code
+        shell: sh
+        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']}/{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:
+              top_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
+              for member in tar.getmembers():
+                  name = member.name
+                  if name == top_prefix[:-1]:
+                      continue
+                  if name.startswith(top_prefix):
+                      member.name = name[len(top_prefix):]
+                  if member.name:
+                      tar.extract(member, '.')
+          PY
+
+      - name: Auto merge develop PRs
+        run: |
+          bash scripts/auto_merge_prs.sh develop
+
+      - name: Auto merge main PRs (release only)
+        run: |
+          bash scripts/auto_merge_prs.sh main
@@ -15,7 +15,6 @@ on:
    branches:
      - main
      - develop
-  workflow_dispatch:

 permissions:
  contents: read
@@ -23,7 +22,7 @@ permissions:
 jobs:
  validate:
    name: Validate Code Quality And Tests
-    runs-on: host
+    runs-on: ubuntu-22.04
    timeout-minutes: 10

    env:
@@ -81,7 +80,7 @@ jobs:
        shell: sh
        run: |
          set -eu
-          python3 --version
+          python --version
          python3 -m pip --version
          echo "CI environment is ready"

@@ -92,7 +91,6 @@ jobs:
          python3 -m pip install -q -r requirements-base.txt
          python3 -m pip install -q -r requirements.txt
          python3 -m pip install -q -r requirements-dev.txt
-          python3 -m pip install -q -r requirements-worker.txt
          python3 -m black --version
          python3 -m isort --version-number
          python3 -m flake8 --version
@@ -167,9 +165,103 @@ jobs:
        run: |
          set -eu
          PYTHONPATH="$PWD/apps/api:$PWD" python3 -m pytest tests/unit -q \
-            --cov=apps/api --cov=apps/common --cov=packages \
-            --cov-report=term --cov-report=term-missing --cov-report=xml \
-            --cov-fail-under=65
+            --cov=apps --cov-report=term --cov-report=term-missing --cov-report=xml \
+            --cov-fail-under=60
+
+      - name: Start PostgreSQL for integration tests
+        shell: sh
+        run: |
+          set -eu
+          PG_CONTAINER="ci-pg-${GITHUB_RUN_ID:-$$}"
+          echo "PG_CONTAINER=$PG_CONTAINER" >> "$GITHUB_ENV"
+          docker rm -f "$PG_CONTAINER" 2>/dev/null || true
+          docker run -d --name "$PG_CONTAINER" \
+            --shm-size=256m \
+            -e POSTGRES_USER=postgres \
+            -e POSTGRES_PASSWORD=postgres \
+            -e POSTGRES_DB=xiaoxia_saas \
+            -P \
+            --health-cmd "pg_isready -U postgres" \
+            --health-interval 5s \
+            --health-timeout 5s \
+            --health-retries 12 \
+            postgres:16
+          # 获取随机映射的端口
+          PG_PORT=$(docker port "$PG_CONTAINER" 5432/tcp | cut -d: -f2)
+          echo "PostgreSQL port: $PG_PORT"
+          echo "DATABASE_URL=postgresql+psycopg://postgres:postgres@127.0.0.1:$PG_PORT/xiaoxia_saas" >> "$GITHUB_ENV"
+          for i in $(seq 1 30); do
+            if docker inspect --format='{{.State.Health.Status}}' "$PG_CONTAINER" 2>/dev/null | grep -q healthy; then
+              echo "PostgreSQL is ready on port $PG_PORT"
+              break
+            fi
+            echo "Waiting for PostgreSQL... ($i/30)"
+            sleep 2
+          done
+          docker inspect --format='{{.State.Health.Status}}' "$PG_CONTAINER" | grep -q healthy
+
+      - name: Apply migrations for integration tests
+        shell: sh
+        run: |
+          set -eu
+          PYTHONPATH="$PWD/apps/api:$PWD" python3 -m alembic upgrade head
+
+      - name: Run integration tests
+        shell: sh
+        run: |
+          set -eu
+          pip install -q pytest-rerunfailures
+          PYTHONPATH="$PWD/apps/api:$PWD" python3 -m pytest tests/integration -q --timeout=60 -x --reruns 2 --reruns-delay 1 -m "not performance" \
+            --cov=apps --cov-append --cov-report=term --cov-report=term-missing --cov-report=xml --cov-fail-under=65
+
+      - name: Run API performance baseline tests
+        shell: sh
+        continue-on-error: true
+        run: |
+          set +e
+          echo "=== API 性能基线测试 ==="
+          PERF_OUTPUT=$(mktemp)
+          PYTHONPATH="$PWD/apps/api:$PWD" python3 -m pytest tests/integration/test_api_performance.py \
+            -v --timeout=120 -p no:cacheprovider 2>&1 | tee "$PERF_OUTPUT"
+          PERF_EXIT=$?
+          
+          # 提取性能统计
+          echo ""
+          echo "=== 性能测试摘要 ==="
+          grep "PERF_STATS:" "$PERF_OUTPUT" || echo "PERF_STATS: 未找到统计数据"
+          grep "PERF_RESULT:" "$PERF_OUTPUT" || echo "PERF_RESULT: 未找到详细结果"
+          
+          # 统计通过率
+          TOTAL=$(grep -c "PERF_RESULT:" "$PERF_OUTPUT" || echo 0)
+          PASSED=$(grep "PERF_RESULT: PASS" "$PERF_OUTPUT" | wc -l)
+          FAILED=$(grep "PERF_RESULT: FAIL" "$PERF_OUTPUT" | wc -l)
+          
+          echo ""
+          echo "性能测试结果: $PASSED/$TOTAL 通过, $FAILED 未达标"
+          
+          if [ "$FAILED" -gt 0 ]; then
+            echo ""
+            echo "⚠️  警告: $FAILED 个接口性能未达标，请关注以下接口:"
+            grep "PERF_RESULT: FAIL" "$PERF_OUTPUT" | while read line; do
+              echo "   $line"
+            done
+            echo ""
+            echo "性能测试失败不阻塞主流水线，但建议尽快优化。"
+          else
+            echo "✅ 所有接口性能达标！"
+          fi
+          
+          rm -f "$PERF_OUTPUT"
+          # 始终返回 0，不阻塞流水线
+          exit 0
+
+
+      - name: Cleanup PostgreSQL
+        if: always()
+        shell: sh
+        run: |
+          docker rm -f "${PG_CONTAINER:-ci-pg-validate}" 2>/dev/null || true
+          echo "PostgreSQL container cleaned up"

      - name: Coverage summary
        if: always()
@@ -177,21 +269,92 @@ jobs:
        run: |
          set +e
          echo "=== 覆盖率汇总 ==="
-          if [ -f .coverage ]; then
-            python3 -m coverage report
-          elif [ -f coverage.xml ]; then
-            python3 -m coverage report --data-file=coverage.xml 2>/dev/null || echo "coverage.xml 存在但无法解析"
+          if [ -f coverage.xml ]; then
+            python3 -c "
+import xml.etree.ElementTree as ET
+tree = ET.parse('coverage.xml')
+root = tree.getroot()
+line_rate = float(root.get('line-rate', 0)) * 100
+branch_rate = float(root.get('branch-rate', 0)) * 100
+lines_covered = int(root.get('lines-covered', 0))
+lines_valid = int(root.get('lines-valid', 0))
+print(f'行覆盖率: {line_rate:.2f}% ({lines_covered}/{lines_valid})')
+print(f'分支覆盖率: {branch_rate:.2f}%')
+print(f'门槛: 65%')
+print(f'状态: {"PASS ✅" if line_rate >= 65 else "FAIL ❌"}')
+"
          else
-            echo "无覆盖率数据"
+            echo "coverage.xml 不存在，跳过汇总"
          fi

-      - name: Notify failure - Validate Code Quality And Tests
+      - name: Notify CI failure
        if: failure()
        shell: sh
-        env:
-          CI_NOTIFY_WEBHOOK: ${{ secrets.CI_NOTIFY_WEBHOOK }}
        run: |
-          bash .gitea/scripts/send_feishu_notify.sh failure "Validate Code Quality And Tests"
+          set +e
+          echo "=== CI 失败通知 ==="
+          
+          # 收集失败信息
+          FAILED_JOB="Validate Code Quality And Tests"
+          BRANCH="${GITHUB_REF_NAME:-unknown}"
+          COMMIT="${GITHUB_SHA:0:8}"
+          ACTOR="${GITHUB_ACTOR:-unknown}"
+          RUN_ID="${GITHUB_RUN_ID:-unknown}"
+          REPO="${GITHUB_REPOSITORY:-unknown}"
+          RUN_URL="https://git.xiaoxiajianji.com/${REPO}/actions/runs/${RUN_ID}"
+          
+          # 构造通知消息
+          PAYLOAD=$(cat <<EOF
+          {
+            "msg_type": "interactive",
+            "card": {
+              "header": {
+                "title": {
+                  "tag": "plain_text",
+                  "content": "❌ CI 构建失败"
+                },
+                "status": "red"
+              },
+              "elements": [
+                {
+                  "tag": "div",
+                  "text": {
+                    "tag": "lark_md",
+                    "content": "**任务**: ${FAILED_JOB}
+**分支**: ${BRANCH}
+**提交**: ${COMMIT}
+**提交者**: ${ACTOR}
+**Run ID**: ${RUN_ID}"
+                  }
+                },
+                {
+                  "tag": "action",
+                  "actions": [
+                    {
+                      "tag": "button",
+                      "text": {
+                        "tag": "plain_text",
+                        "content": "查看失败日志"
+                      },
+                      "url": "${RUN_URL}",
+                      "type": "danger"
+                    }
+                  ]
+                }
+              ]
+            }
+          }
+          EOF
+          )
+          
+          # 如果配置了通知 webhook 就发送
+          if [ -n "${CI_NOTIFY_WEBHOOK:-}" ]; then
+            curl -s -X POST -H "Content-Type: application/json"               "${CI_NOTIFY_WEBHOOK}"               -d "$PAYLOAD" > /dev/null 2>&1 && echo "通知已发送" || echo "通知发送失败"
+          else
+            echo "未配置 CI_NOTIFY_WEBHOOK，跳过通知"
+            echo "如需启用，请在仓库 Settings -> Secrets and variables -> Actions 中添加 CI_NOTIFY_WEBHOOK"
+          fi
+
      - name: Build summary
        if: github.ref == 'refs/heads/develop' || github.ref == 'refs/heads/main'
        shell: sh
@@ -201,218 +364,19 @@ jobs:
          echo "Branch: ${GITHUB_REF_NAME}"
          echo "Commit: ${GITHUB_SHA}"
          # 输出最终覆盖率
-          if [ -f .coverage ]; then
-            python3 -m coverage report | tail -1
+          if [ -f coverage.xml ]; then
+            python3 -c "
+import xml.etree.ElementTree as ET
+tree = ET.parse('coverage.xml')
+root = tree.getroot()
+line_rate = float(root.get('line-rate', 0)) * 100
+print(f'Total coverage: {line_rate:.2f}%')
+"
          fi

-
-  integration-tests:
-    name: Integration Tests
-    runs-on: host
-    timeout-minutes: 20
-
-    steps:
-    - name: Checkout code
-      shell: sh
-      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
-
-    - name: Verify CI environment
-      shell: sh
-      run: |
-        set -eu
-        python3 --version
-        python3 -m pip --version
-        echo "CI environment is ready"
-
-    - name: Install dependencies
-      shell: sh
-      run: |
-        set -eu
-        python3 -m pip install -q -r requirements-base.txt
-        python3 -m pip install -q -r requirements.txt
-        python3 -m pip install -q -r requirements-dev.txt
-        python3 -m pip install -q -r requirements-worker.txt
-        pytest --version
-
-    - name: Start PostgreSQL
-      shell: sh
-      run: |
-        set -eu
-        PG_CONTAINER="ci-pg-int-${GITHUB_RUN_ID:-$$}"
-        echo "PG_CONTAINER=$PG_CONTAINER" >> "$GITHUB_ENV"
-        docker rm -f "$PG_CONTAINER" 2>/dev/null || true
-        docker run -d --name "$PG_CONTAINER" \
-          --shm-size=256m \
-          -e POSTGRES_USER=postgres \
-          -e POSTGRES_PASSWORD=postgres \
-          -e POSTGRES_DB=xiaoxia_saas \
-          -P \
-          --health-cmd "pg_isready -U postgres" \
-          --health-interval 5s \
-          --health-timeout 5s \
-          --health-retries 12 \
-          postgres:16
-        # 获取随机映射的端口
-        PG_PORT=$(docker port "$PG_CONTAINER" 5432/tcp | cut -d: -f2)
-        echo "PostgreSQL port: $PG_PORT"
-        echo "DATABASE_URL=postgresql+psycopg://postgres:postgres@127.0.0.1:$PG_PORT/xiaoxia_saas" >> "$GITHUB_ENV"
-        for i in $(seq 1 30); do
-          if docker inspect --format='{{.State.Health.Status}}' "$PG_CONTAINER" 2>/dev/null | grep -q healthy; then
-            echo "PostgreSQL is ready on port $PG_PORT"
-            break
-          fi
-          echo "Waiting for PostgreSQL... ($i/30)"
-          sleep 2
-        done
-        docker inspect --format='{{.State.Health.Status}}' "$PG_CONTAINER" | grep -q healthy
-
-    - name: Start Redis
-      shell: sh
-      run: |
-        set -eu
-        REDIS_CONTAINER="ci-redis-int-${GITHUB_RUN_ID:-$$}"
-        echo "REDIS_CONTAINER=$REDIS_CONTAINER" >> "$GITHUB_ENV"
-        docker rm -f "$REDIS_CONTAINER" 2>/dev/null || true
-        docker run -d --name "$REDIS_CONTAINER"           -P           --health-cmd "redis-cli ping"           --health-interval 2s           --health-timeout 2s           --health-retries 10           redis:7-alpine
-        REDIS_PORT=$(docker port "$REDIS_CONTAINER" 6379/tcp | cut -d: -f2)
-        echo "Redis port: $REDIS_PORT"
-        echo "REDIS_URL=redis://127.0.0.1:$REDIS_PORT/0" >> "$GITHUB_ENV"
-        for i in $(seq 1 15); do
-          if docker inspect --format='{{.State.Health.Status}}' "$REDIS_CONTAINER" 2>/dev/null | grep -q healthy; then
-            echo "Redis is ready on port $REDIS_PORT"
-            break
-          fi
-          echo "Waiting for Redis... ($i/15)"
-          sleep 2
-        done
-        docker inspect --format='{{.State.Health.Status}}' "$REDIS_CONTAINER" | grep -q healthy
-
-    - name: Apply migrations
-      shell: sh
-      run: |
-        set -eu
-        PYTHONPATH="$PWD/apps/api:$PWD" python3 -m alembic upgrade head
-
-    - name: Run integration tests
-      shell: sh
-      run: |
-        set -eu
-        pip install -q pytest-rerunfailures
-        PYTHONPATH="$PWD/apps/api:$PWD" python3 -m pytest tests/integration -q --timeout=60 -x --reruns 2 --reruns-delay 1 -m "not performance" \
-          --cov=apps --cov-report=term --cov-report=term-missing --cov-report=xml --cov-fail-under=50
-
-    - name: Run API performance baseline tests
-      shell: sh
-      continue-on-error: true
-      run: |
-        set +e
-        echo "=== API 性能基线测试 ==="
-        PERF_OUTPUT=$(mktemp)
-        PYTHONPATH="$PWD/apps/api:$PWD" python3 -m pytest tests/integration/test_api_performance.py \
-          -v --timeout=120 -p no:cacheprovider 2>&1 | tee "$PERF_OUTPUT"
-        PERF_EXIT=$?
-        
-        # 提取性能统计
-        echo ""
-        echo "=== 性能测试摘要 ==="
-        grep "PERF_STATS:" "$PERF_OUTPUT" || echo "PERF_STATS: 未找到统计数据"
-        grep "PERF_RESULT:" "$PERF_OUTPUT" || echo "PERF_RESULT: 未找到详细结果"
-        
-        # 统计通过率
-        TOTAL=$(grep -c "PERF_RESULT:" "$PERF_OUTPUT" || echo 0)
-        PASSED=$(grep "PERF_RESULT: PASS" "$PERF_OUTPUT" | wc -l)
-        FAILED=$(grep "PERF_RESULT: FAIL" "$PERF_OUTPUT" | wc -l)
-        
-        echo ""
-        echo "性能测试结果: $PASSED/$TOTAL 通过, $FAILED 未达标"
-        
-        if [ "$FAILED" -gt 0 ]; then
-          echo ""
-          echo "⚠️  警告: $FAILED 个接口性能未达标，请关注以下接口:"
-          grep "PERF_RESULT: FAIL" "$PERF_OUTPUT" | while read line; do
-            echo "   $line"
-          done
-          echo ""
-          echo "性能测试失败不阻塞主流水线，但建议尽快优化。"
-        else
-          echo "✅ 所有接口性能达标！"
-        fi
-        
-        rm -f "$PERF_OUTPUT"
-        # 始终返回 0，不阻塞流水线
-        exit 0
-
-    - name: Cleanup PostgreSQL
-      if: always()
-      shell: sh
-      run: |
-        docker rm -f "${PG_CONTAINER:-ci-pg-int}" 2>/dev/null || true
-        echo "PostgreSQL container cleaned up"
-
-    - name: Coverage summary
-      if: always()
-      shell: sh
-      run: |
-        set +e
-        echo "=== 集成测试覆盖率汇总 ==="
-        if [ -f .coverage ]; then
-          python3 -m coverage report
-        else
-          echo "无覆盖率数据"
-        fi
-
-    - name: Notify failure - Integration Tests
-      if: failure()
-      shell: sh
-      env:
-        CI_NOTIFY_WEBHOOK: ${{ secrets.CI_NOTIFY_WEBHOOK }}
-      run: |
-        bash .gitea/scripts/send_feishu_notify.sh failure "Integration Tests"
  frontend-lint:
    name: Frontend Lint
-    runs-on: host
+    runs-on: ubuntu-22.04
    timeout-minutes: 10

    steps:
@@ -511,17 +475,9 @@ jobs:
            -w /workspace/apps/web \
            docker.m.daocloud.io/library/node:20 \
            sh -lc 'npx vitest run src/test'
-
-      - name: Notify failure - Frontend Lint
-        if: failure()
-        shell: sh
-        env:
-          CI_NOTIFY_WEBHOOK: ${{ secrets.CI_NOTIFY_WEBHOOK }}
-        run: |
-          bash .gitea/scripts/send_feishu_notify.sh failure "Frontend Lint"
  deploy-staging:
    name: Build & Push Staging (Watchtower auto-deploy)
-    runs-on: host
+    runs-on: saas
    timeout-minutes: 30
    needs: [validate, frontend-lint]

@@ -651,23 +607,9 @@ jobs:
        echo "Commit: ${GITHUB_SHA}"


-    - name: Notify failure - Deploy Staging
-      if: failure()
-      shell: sh
-      env:
-        CI_NOTIFY_WEBHOOK: ${{ secrets.CI_NOTIFY_WEBHOOK }}
-      run: |
-        bash .gitea/scripts/send_feishu_notify.sh failure "Deploy Staging"
-    - name: Notify success - Staging 镜像就绪
-      if: success() && (github.ref_name == 'develop' || github.ref_name == 'main')
-      shell: sh
-      env:
-        CI_NOTIFY_WEBHOOK: ${{ secrets.CI_NOTIFY_WEBHOOK }}
-      run: |
-        bash .gitea/scripts/send_feishu_notify.sh success "Staging 镜像就绪"
  staging-e2e:
    name: Staging E2E Tests
-    runs-on: host
+    runs-on: saas
    timeout-minutes: 15
    if: github.ref_name == 'develop' || github.ref_name == 'main'
    needs: deploy-staging
@@ -734,7 +676,7 @@ jobs:

  staging-api-tests:
    name: Staging API Integration Tests
-    runs-on: host
+    runs-on: saas
    timeout-minutes: 10
    if: github.ref_name == 'develop' || github.ref_name == 'main'
    needs: deploy-staging
@@ -800,7 +742,7 @@ jobs:

  build-production-runtime-images:
    name: Build Production Runtime Images
-    runs-on: host
+    runs-on: saas
    timeout-minutes: 30
    needs: [validate, frontend-lint]

@@ -879,7 +821,7 @@ jobs:

  deploy-production:
    name: Deploy Production
-    runs-on: host
+    runs-on: saas
    timeout-minutes: 20
    if: startsWith(github.ref, 'refs/tags/v')
    needs: build-production-runtime-images
@@ -944,7 +886,7 @@ jobs:

  production-e2e:
    name: Production Browser E2E
-    runs-on: host
+    runs-on: saas
    timeout-minutes: 15
    if: startsWith(github.ref, 'refs/tags/v')
    needs: deploy-production
@@ -0,0 +1,69 @@
+name: Test SSH Secret
+on:
+  push:
+    branches: [develop]
+    paths:
+      - '.gitea/workflows/test-ssh-secret.yml'
+
+jobs:
+  test-ssh:
+    runs-on: ubuntu-22.04
+    steps:
+      - name: Install SSH client
+        run: |
+          which ssh || (apt-get update && apt-get install -y openssh-client)
+          ssh -V
+
+      - name: Debug environment
+        run: |
+          echo "=== Environment ==="
+          echo "Runner hostname: $(hostname)"
+          echo "Runner IP: $(hostname -i || echo 'unknown')"
+          echo "Current user: $(whoami)"
+          echo "=== Secrets check ==="
+          if [ -n "$STAGING_SSH_HOST" ]; then
+            echo "STAGING_SSH_HOST: [SET] value_length=${#STAGING_SSH_HOST}"
+          else
+            echo "STAGING_SSH_HOST: [EMPTY]"
+          fi
+          if [ -n "$STAGING_SSH_USER" ]; then
+            echo "STAGING_SSH_USER: [SET] value_length=${#STAGING_SSH_USER}"
+          else
+            echo "STAGING_SSH_USER: [EMPTY]"
+          fi
+          if [ -n "$STAGING_SSH_KEY" ]; then
+            echo "STAGING_SSH_KEY: [SET] value_length=${#STAGING_SSH_KEY}"
+          else
+            echo "STAGING_SSH_KEY: [EMPTY]"
+          fi
+        env:
+          STAGING_SSH_HOST: ${{ secrets.STAGING_SSH_HOST }}
+          STAGING_SSH_USER: ${{ secrets.STAGING_SSH_USER }}
+          STAGING_SSH_KEY: ${{ secrets.STAGING_SSH_KEY }}
+
+      - name: Setup SSH key
+        run: |
+          mkdir -p ~/.ssh
+          chmod 700 ~/.ssh
+          echo "$STAGING_SSH_KEY" > ~/.ssh/id_ed25519
+          chmod 600 ~/.ssh/id_ed25519
+          ssh-keygen -y -f ~/.ssh/id_ed25519 > ~/.ssh/id_ed25519.pub 2>/dev/null || echo "No public key generated"
+          echo "=== SSH Key fingerprint ==="
+          ssh-keygen -lf ~/.ssh/id_ed25519 || echo "Key fingerprint failed"
+        env:
+          STAGING_SSH_KEY: ${{ secrets.STAGING_SSH_KEY }}
+
+      - name: Test SSH connection
+        run: |
+          echo "Attempting SSH connection to $STAGING_SSH_HOST..."
+          ssh -i ~/.ssh/id_ed25519 \
+            -o StrictHostKeyChecking=no \
+            -o UserKnownHostsFile=/dev/null \
+            -o ConnectTimeout=10 \
+            -o BatchMode=yes \
+            -v \
+            $STAGING_SSH_USER@$STAGING_SSH_HOST "echo 'SSH_CONNECTION_SUCCESS' && hostname && whoami"
+          echo "=== SSH Test Complete ==="
+        env:
+          STAGING_SSH_HOST: ${{ secrets.STAGING_SSH_HOST }}
+          STAGING_SSH_USER: ${{ secrets.STAGING_SSH_USER }}
@@ -0,0 +1,163 @@
+name: Tests
+
+on:
+  pull_request:
+    branches: [ main ]
+
+jobs:
+  test:
+    runs-on: runtime-builder
+
+    steps:
+    - name: Checkout code
+      shell: sh
+      run: |
+        set -eu
+        python - <<'PY'
+        import io
+        import os
+        import tarfile
+        import time
+        import urllib.error
+        import urllib.request
+
+        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']}"})
+        # Retry up to 5 times with backoff for transient 5xx errors
+        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
+
+    - name: Show Python version
+      shell: sh
+      run: |
+        set -eu
+        python --version
+        python -m pip --version
+
+    - name: Install dependencies
+      shell: sh
+      run: |
+        set -eu
+        python -m pip install --upgrade pip -i https://mirrors.aliyun.com/pypi/simple/ --trusted-host mirrors.aliyun.com
+        python -m pip install -r requirements.txt -r requirements-dev.txt -i https://mirrors.aliyun.com/pypi/simple/ --trusted-host mirrors.aliyun.com
+
+    - name: Run unit tests
+      shell: sh
+      run: |
+        set -eu
+        PYTHONPATH="$PWD/apps/api:$PWD" python -m pytest tests/unit -q
+
+    - name: Run integration tests
+      shell: sh
+      run: |
+        set -eu
+        PYTHONPATH="$PWD/apps/api:$PWD" python -m pytest tests/integration -q --timeout=60 -x
+
+  lint:
+    runs-on: runtime-builder
+
+    steps:
+    - name: Checkout code
+      shell: sh
+      run: |
+        set -eu
+        python - <<'PY'
+        import io
+        import os
+        import tarfile
+        import time
+        import urllib.error
+        import urllib.request
+
+        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']}"})
+        # Retry up to 5 times with backoff for transient 5xx errors
+        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
+
+    - name: Install dependencies
+      shell: sh
+      run: |
+        set -eu
+        python -m pip install --upgrade pip -i https://mirrors.aliyun.com/pypi/simple/ --trusted-host mirrors.aliyun.com
+        python -m pip install -r requirements.txt -r requirements-dev.txt -i https://mirrors.aliyun.com/pypi/simple/ --trusted-host mirrors.aliyun.com
+
+    - name: Run Black (check only)
+      shell: sh
+      run: |
+        set -eu
+        python -m black --check alembic apps packages tests scripts
+
+    - name: Run Flake8
+      shell: sh
+      run: |
+        set -eu
+        python -m flake8 apps packages tests --count --statistics
@@ -6,8 +6,8 @@ from app.core.celery_app import celery_app
 logger = logging.getLogger(__name__)

 # ── 限流阈值常量（全系统统一管理，不要在业务代码里硬编码） ──
-USER_PENDING_LIMIT = 3  # 单用户 pending 上限
-GLOBAL_PENDING_LIMIT = 20  # 全局 pending 上限
+USER_PENDING_LIMIT = 3       # 单用户 pending 上限
+GLOBAL_PENDING_LIMIT = 20    # 全局 pending 上限


 class UserPendingLimitExceeded(Exception):
@@ -71,7 +71,9 @@ def check_queue_limits(
                user_pending,
                user_pending_limit,
            )
-            raise UserPendingLimitExceeded(user_id=user_id, pending_count=user_pending, limit=user_pending_limit)
+            raise UserPendingLimitExceeded(
+                user_id=user_id, pending_count=user_pending, limit=user_pending_limit
+            )


 def _mark_task_failed_safely(
@@ -155,7 +157,9 @@ def safe_enqueue_generation_task(
                user_pending,
                user_pending_limit,
            )
-            exc = UserPendingLimitExceeded(user_id=user_id, pending_count=user_pending, limit=user_pending_limit)
+            exc = UserPendingLimitExceeded(
+                user_id=user_id, pending_count=user_pending, limit=user_pending_limit
+            )
            _mark_task_failed_safely(task, generation_task_repository, log_prefix, str(exc))
            raise exc

@@ -197,7 +201,9 @@ def safe_enqueue_generation_task(
            exc: Exception = GlobalQueueFull(pending_count=global_after, limit=global_pending_limit)
        else:
            reason = f"用户 pending 超限(入队后): {user_after}/{user_pending_limit}"
-            exc = UserPendingLimitExceeded(user_id=user_id, pending_count=user_after, limit=user_pending_limit)
+            exc = UserPendingLimitExceeded(
+                user_id=user_id, pending_count=user_after, limit=user_pending_limit
+            )

        logger.warning(
            "[队列限流] %s, task_id=%s, user_id=%s — 回滚状态为 failed",
@@ -118,16 +118,6 @@ class StubGenerationTaskRepository:
    def count_by_user(self, user_id: str) -> int:
        return len([t for t in self._tasks.values() if t.created_by_user_id == user_id])

-    def count_pending_by_user(self, user_id: str) -> int:
-        from app.models.generation_task import TaskStatus
-        pending_statuses = {TaskStatus.PENDING, TaskStatus.PROCESSING, TaskStatus.QUEUED}
-        return len([t for t in self._tasks.values() if t.created_by_user_id == user_id and t.status in pending_statuses])
-
-    def count_pending_total(self) -> int:
-        from app.models.generation_task import TaskStatus
-        pending_statuses = {TaskStatus.PENDING, TaskStatus.PROCESSING, TaskStatus.QUEUED}
-        return len([t for t in self._tasks.values() if t.status in pending_statuses])
-
    def list_recent_by_user(self, user_id: str, limit: int = 5) -> list[GenerationTask]:
        items = [t for t in self._tasks.values() if t.created_by_user_id == user_id]
        items.sort(key=lambda t: t.created_at, reverse=True)
@@ -1,9 +1,8 @@
 """任务队列限流防护单元测试。"""
-
 from __future__ import annotations

-import os
 import sys
+import os
 from unittest.mock import MagicMock

 import pytest
@@ -19,6 +18,7 @@ from app.core.task_enqueue import (
    safe_enqueue_generation_task,
 )

+
 # ---------------------------------------------------------------------------
 # Mock helpers
 # ---------------------------------------------------------------------------