diff --git a/scripts/ci/ci_dashboard.py b/scripts/ci/ci_dashboard.py
new file mode 100644
index 000000000..a8bcaac5c
--- /dev/null
+++ b/scripts/ci/ci_dashboard.py
@@ -0,0 +1,575 @@
+#!/usr/bin/env python3
+"""
+CI 可观测性看板 - 从 Gitea Actions API 拉取数据并生成 Markdown 日报
+
+用法:
+  python3 scripts/ci/ci_dashboard.py --days 7
+  python3 scripts/ci/ci_dashboard.py --days 30 --output ci_report.md
+  python3 scripts/ci/ci_dashboard.py --workflow ci-cd.yml --days 7
+
+环境变量:
+  GITEA_URL      Gitea 地址 (默认 https://git.xiaoxiajianji.com)
+  GITEA_REPO     仓库 (默认 xiaoxia/xiaoxia-saas)
+  GITEA_TOKEN    API Token (优先) 或 GITEA_USERNAME + GITEA_PASSWORD
+"""
+
+import argparse
+import base64
+import json
+import math
+import os
+import statistics
+import sys
+import urllib.error
+import urllib.request
+from collections import defaultdict
+from datetime import datetime, timedelta, timezone
+
+# ── 配置 ──────────────────────────────────────────────
+DEFAULT_GITEA_URL = "https://git.xiaoxiajianji.com"
+DEFAULT_REPO = "xiaoxia/xiaoxia-saas"
+DEFAULT_DAYS = 7
+PAGE_LIMIT = 50  # 每页数量，最大50
+
+
+# ── API 封装 ─────────────────────────────────────────
+class GiteaActions:
+    def __init__(self, base_url, repo, token=None, username=None, password=None):
+        self.base_url = base_url.rstrip("/")
+        self.repo = repo
+        self.token = token
+        self.username = username
+        self.password = password
+        self.api_base = f"{self.base_url}/api/v1/repos/{self.repo}/actions"
+
+    def _request(self, path):
+        url = f"{self.api_base}/{path}"
+        req = urllib.request.Request(url)
+        if self.token:
+            req.add_header("Authorization", f"token {self.token}")
+        elif self.username and self.password:
+            auth = base64.b64encode(f"{self.username}:{self.password}".encode()).decode()
+            req.add_header("Authorization", f"Basic {auth}")
+        try:
+            with urllib.request.urlopen(req, timeout=30) as resp:
+                return json.loads(resp.read().decode())
+        except urllib.error.HTTPError as e:
+            print(f"[WARN] HTTP {e.code}: {url}", file=sys.stderr)
+            return None
+        except Exception as e:
+            print(f"[WARN] 请求失败 {url}: {e}", file=sys.stderr)
+            return None
+
+    def list_runs(self, status=None, branch=None, event=None, page=1, limit=PAGE_LIMIT):
+        """获取 workflow runs 列表"""
+        params = []
+        if status:
+            params.append(f"status={status}")
+        if branch:
+            params.append(f"branch={branch}")
+        if event:
+            params.append(f"event={event}")
+        params.append(f"page={page}")
+        params.append(f"limit={limit}")
+        path = f"runs?{'&'.join(params)}"
+        data = self._request(path)
+        if not data:
+            return [], 0
+        runs = data.get("workflow_runs", [])
+        total = data.get("total_count", 0)
+        return runs, total
+
+    def get_run_jobs(self, run_id):
+        """获取 run 的所有 job"""
+        data = self._request(f"runs/{run_id}/jobs")
+        if not data:
+            return []
+        return data.get("jobs", [])
+
+    def list_workflows(self):
+        """获取所有 workflow"""
+        data = self._request("workflows")
+        if not data:
+            return []
+        return data.get("workflows", [])
+
+
+# ── 工具函数 ─────────────────────────────────────────
+def parse_datetime(s):
+    """解析 ISO 格式时间字符串"""
+    if not s or s.startswith("1970") or s.startswith("0001"):
+        return None
+    try:
+        if s.endswith("Z"):
+            s = s[:-1] + "+00:00"
+        return datetime.fromisoformat(s)
+    except Exception:
+        return None
+
+
+def to_shanghai(dt):
+    """转换为上海时区"""
+    if dt is None:
+        return None
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    return dt.astimezone(timezone(timedelta(hours=8)))
+
+
+def duration_seconds(start_str, end_str):
+    """计算耗时（秒）"""
+    start = parse_datetime(start_str)
+    end = parse_datetime(end_str)
+    if not start or not end:
+        return None
+    return (end - start).total_seconds()
+
+
+def fmt_duration(seconds):
+    """格式化耗时显示"""
+    if seconds is None:
+        return "N/A"
+    seconds = int(seconds)
+    if seconds < 60:
+        return f"{seconds}s"
+    mins, secs = divmod(seconds, 60)
+    if mins < 60:
+        return f"{mins}m{secs:02d}s"
+    hours, mins = divmod(mins, 60)
+    return f"{hours}h{mins:02d}m"
+
+
+def percentile(sorted_values, p):
+    """计算百分位数"""
+    if not sorted_values:
+        return None
+    k = (len(sorted_values) - 1) * (p / 100)
+    f = math.floor(k)
+    c = math.ceil(k)
+    if f == c:
+        return sorted_values[int(k)]
+    return sorted_values[f] * (c - k) + sorted_values[c] * (k - f)
+
+
+def classify_failure(job_name, step_name=None):
+    """根据失败的 job/step 名称分类失败原因"""
+    name = f"{job_name} {step_name or ''}".lower()
+    if any(k in name for k in ["lint", "ruff", "flake8", "eslint", "prettier", "black", "mypy"]):
+        return "代码质量 / Lint"
+    if any(k in name for k in ["unit test", "pytest", "vitest", "jest"]):
+        return "单元测试失败"
+    if any(k in name for k in ["integration", "e2e"]):
+        return "集成测试 / E2E"
+    if any(k in name for k in ["build", "compile", "docker", "image"]):
+        return "构建失败"
+    if any(k in name for k in ["deploy", "preview", "release"]):
+        return "部署失败"
+    if any(k in name for k in ["setup", "checkout", "cache", "install", "deps"]):
+        return "环境 / 依赖"
+    if any(k in name for k in ["migrate", "migration", "schema"]):
+        return "数据库迁移"
+    return "其他"
+
+
+# ── 数据收集 ─────────────────────────────────────────
+def fetch_runs_in_range(ga, start_date, end_date, workflow_filter=None):
+    """拉取指定日期范围内的所有 completed runs"""
+    all_runs = []
+    page = 1
+    print(f"[INFO] 拉取 {start_date} ~ {end_date} 的 CI runs...", file=sys.stderr)
+
+    while True:
+        runs, total = ga.list_runs(status="completed", page=page, limit=PAGE_LIMIT)
+        if not runs:
+            break
+
+        if workflow_filter:
+            runs = [r for r in runs if workflow_filter in r.get("path", "")]
+
+        in_range = []
+        out_range_old = False
+
+        for run in runs:
+            started = to_shanghai(parse_datetime(run.get("started_at")))
+            if not started:
+                continue
+            run_date = started.date()
+            if start_date <= run_date <= end_date:
+                in_range.append(run)
+            elif run_date < start_date:
+                out_range_old = True
+
+        all_runs.extend(in_range)
+        print(
+            f"[INFO] 第 {page} 页: {len(runs)} 条, 范围内 {len(in_range)} 条, 累计 {len(all_runs)} 条", file=sys.stderr
+        )
+
+        if out_range_old or len(runs) < PAGE_LIMIT:
+            break
+
+        page += 1
+        if page > 100:
+            print("[WARN] 超过100页，停止拉取", file=sys.stderr)
+            break
+
+    print(f"[INFO] 共获取 {len(all_runs)} 条 run 数据", file=sys.stderr)
+    return all_runs
+
+
+def enrich_with_jobs(ga, runs, max_failures=50):
+    """为 runs 补充 job 详情（失败原因分析 + runner 统计）
+
+    失败 run 按时间倒序取最近 N 个（避免 API 调用过多），
+    成功 run 采样用于 runner 分布统计。
+    """
+    # 失败 run 取最近 N 个
+    failure_runs = [r for r in runs if r.get("conclusion") != "success"]
+    failure_runs = failure_runs[:max_failures]  # 已经是时间倒序
+    print(f"[INFO] 为最近 {len(failure_runs)} 个失败 run 拉取 job 详情...", file=sys.stderr)
+
+    for i, run in enumerate(failure_runs):
+        jobs = ga.get_run_jobs(run["id"])
+        run["_jobs"] = jobs
+        if (i + 1) % 10 == 0:
+            print(f"[INFO] 已处理 {i+1}/{len(failure_runs)}", file=sys.stderr)
+
+    # 成功 run 采样用于 runner 分布
+    success_runs = [r for r in runs if r.get("conclusion") == "success"]
+    sample_size = min(50, len(success_runs))
+    if sample_size > 0:
+        sampled = success_runs[:: max(1, len(success_runs) // sample_size)]
+        print(f"[INFO] 采样 {len(sampled)} 个成功 run 用于 runner 统计...", file=sys.stderr)
+        for run in sampled:
+            if "_jobs" not in run:
+                jobs = ga.get_run_jobs(run["id"])
+                run["_jobs"] = jobs
+
+    return runs
+
+
+# ── 统计分析 ─────────────────────────────────────────
+def analyze_runs(runs):
+    """对 runs 做全面统计分析"""
+    if not runs:
+        return {}
+
+    # 基础统计
+    total = len(runs)
+    success = sum(1 for r in runs if r.get("conclusion") == "success")
+    failure = sum(1 for r in runs if r.get("conclusion") == "failure")
+    cancelled = sum(1 for r in runs if r.get("conclusion") == "cancelled")
+    other = total - success - failure - cancelled
+    success_rate = (success / total * 100) if total > 0 else 0
+
+    # 耗时统计
+    durations = []
+    for r in runs:
+        d = duration_seconds(r.get("started_at"), r.get("completed_at"))
+        if d and d > 0:
+            durations.append(d)
+    durations.sort()
+
+    avg_dur = statistics.mean(durations) if durations else None
+    median_dur = percentile(durations, 50)
+    p95_dur = percentile(durations, 95)
+
+    # 按日期统计
+    daily_stats = defaultdict(lambda: {"total": 0, "success": 0, "failure": 0, "durations": []})
+    for r in runs:
+        started = to_shanghai(parse_datetime(r.get("started_at")))
+        if not started:
+            continue
+        day = started.date().isoformat()
+        daily_stats[day]["total"] += 1
+        if r.get("conclusion") == "success":
+            daily_stats[day]["success"] += 1
+        elif r.get("conclusion") == "failure":
+            daily_stats[day]["failure"] += 1
+        d = duration_seconds(r.get("started_at"), r.get("completed_at"))
+        if d and d > 0:
+            daily_stats[day]["durations"].append(d)
+
+    # 按 workflow 统计
+    wf_stats = defaultdict(lambda: {"total": 0, "success": 0, "failure": 0, "durations": []})
+    for r in runs:
+        path = r.get("path", "")
+        wf_name = path.split("@")[0] if "@" in path else path
+        wf_stats[wf_name]["total"] += 1
+        if r.get("conclusion") == "success":
+            wf_stats[wf_name]["success"] += 1
+        elif r.get("conclusion") == "failure":
+            wf_stats[wf_name]["failure"] += 1
+        d = duration_seconds(r.get("started_at"), r.get("completed_at"))
+        if d and d > 0:
+            wf_stats[wf_name]["durations"].append(d)
+
+    # 按触发事件统计
+    event_stats = defaultdict(lambda: {"total": 0, "success": 0, "failure": 0})
+    for r in runs:
+        evt = r.get("event", "unknown")
+        event_stats[evt]["total"] += 1
+        if r.get("conclusion") == "success":
+            event_stats[evt]["success"] += 1
+        elif r.get("conclusion") == "failure":
+            event_stats[evt]["failure"] += 1
+
+    # 失败原因 + runner + job 耗时（需要 _jobs 数据）
+    failure_categories = defaultdict(int)
+    failed_jobs_by_name = defaultdict(int)
+    runner_stats = defaultdict(lambda: {"jobs": 0, "success": 0, "failure": 0, "durations": []})
+    job_time_stats = defaultdict(list)
+
+    for r in runs:
+        jobs = r.get("_jobs", [])
+        if not jobs:
+            continue
+        for job in jobs:
+            runner = job.get("runner_name", "unknown")
+            conclusion = job.get("conclusion", "unknown")
+            runner_stats[runner]["jobs"] += 1
+            if conclusion == "success":
+                runner_stats[runner]["success"] += 1
+            elif conclusion == "failure":
+                runner_stats[runner]["failure"] += 1
+
+            jd = duration_seconds(job.get("started_at"), job.get("completed_at"))
+            if jd and jd > 0:
+                runner_stats[runner]["durations"].append(jd)
+                job_time_stats[job.get("name", "unknown")].append(jd)
+
+            if conclusion == "failure":
+                failed_jobs_by_name[job.get("name", "unknown")] += 1
+                failed_step = None
+                for step in job.get("steps", []):
+                    if step.get("conclusion") == "failure":
+                        failed_step = step.get("name")
+                        break
+                category = classify_failure(job.get("name", ""), failed_step)
+                failure_categories[category] += 1
+
+    return {
+        "total": total,
+        "success": success,
+        "failure": failure,
+        "cancelled": cancelled,
+        "other": other,
+        "success_rate": success_rate,
+        "avg_duration": avg_dur,
+        "median_duration": median_dur,
+        "p95_duration": p95_dur,
+        "durations": durations,
+        "daily_stats": dict(sorted(daily_stats.items())),
+        "workflow_stats": dict(wf_stats),
+        "event_stats": dict(event_stats),
+        "failure_categories": dict(failure_categories),
+        "failed_jobs_top": dict(sorted(failed_jobs_by_name.items(), key=lambda x: -x[1])[:15]),
+        "runner_stats": dict(runner_stats),
+        "job_time_stats": dict(job_time_stats),
+    }
+
+
+# ── 报表生成 ─────────────────────────────────────────
+def generate_markdown(stats, start_date, end_date, repo):
+    """生成 Markdown 格式的日报"""
+    lines = []
+
+    lines.append("# CI 运行状态看板")
+    lines.append("")
+    lines.append(f"> 统计周期: **{start_date} ~ {end_date}**")
+    lines.append(f"> 仓库: `{repo}`")
+    lines.append(f"> 生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+    lines.append("")
+
+    # 概览
+    lines.append("## 📊 整体概览")
+    lines.append("")
+    lines.append("| 指标 | 数值 |")
+    lines.append("|------|------|")
+    lines.append(f"| 总构建次数 | **{stats['total']}** |")
+    lines.append(f"| ✅ 成功 | {stats['success']} |")
+    lines.append(f"| ❌ 失败 | {stats['failure']} |")
+    lines.append(f"| ⏹️ 取消 | {stats['cancelled']} |")
+    lines.append(f"| 📈 成功率 | **{stats['success_rate']:.1f}%** |")
+    lines.append(f"| ⏱️ 平均耗时 | {fmt_duration(stats['avg_duration'])} |")
+    lines.append(f"| ⏱️ P50 耗时 | {fmt_duration(stats['median_duration'])} |")
+    lines.append(f"| ⏱️ P95 耗时 | {fmt_duration(stats['p95_duration'])} |")
+    lines.append("")
+
+    # 每日趋势
+    lines.append("## 📈 每日趋势")
+    lines.append("")
+    lines.append("| 日期 | 总次数 | 成功 | 失败 | 成功率 | 平均耗时 | P95 耗时 |")
+    lines.append("|------|--------|------|------|--------|----------|----------|")
+    for day, s in stats["daily_stats"].items():
+        rate = (s["success"] / s["total"] * 100) if s["total"] > 0 else 0
+        durations = sorted(s["durations"])
+        avg = statistics.mean(durations) if durations else None
+        p95 = percentile(durations, 95) if durations else None
+        lines.append(
+            f"| {day} | {s['total']} | {s['success']} | {s['failure']} | {rate:.1f}% | {fmt_duration(avg)} | {fmt_duration(p95)} |"
+        )
+    lines.append("")
+
+    # 成功率趋势图
+    lines.append("### 成功率趋势图")
+    lines.append("")
+    lines.append("```")
+    max_bar = 40
+    days = list(stats["daily_stats"].keys())
+    if len(days) > 14:
+        days = days[-14:]
+    for day in days:
+        s = stats["daily_stats"][day]
+        rate = (s["success"] / s["total"] * 100) if s["total"] > 0 else 0
+        bar_len = int(rate / 100 * max_bar)
+        bar = "█" * bar_len + "░" * (max_bar - bar_len)
+        lines.append(f"{day} {bar} {rate:5.1f}% ({s['total']}次)")
+    lines.append("```")
+    lines.append("")
+
+    # 按 Workflow 统计
+    lines.append("## 🧩 各 Workflow 统计")
+    lines.append("")
+    wf_sorted = sorted(stats["workflow_stats"].items(), key=lambda x: -x[1]["total"])
+    lines.append("| Workflow | 次数 | 成功 | 失败 | 成功率 | 平均耗时 | P95 耗时 |")
+    lines.append("|----------|------|------|------|--------|----------|----------|")
+    for wf, s in wf_sorted:
+        rate = (s["success"] / s["total"] * 100) if s["total"] > 0 else 0
+        durations = sorted(s["durations"])
+        avg = statistics.mean(durations) if durations else None
+        p95 = percentile(durations, 95) if durations else None
+        wf_short = wf.split("/")[-1] if "/" in wf else wf
+        lines.append(
+            f"| `{wf_short}` | {s['total']} | {s['success']} | {s['failure']} | {rate:.1f}% | {fmt_duration(avg)} | {fmt_duration(p95)} |"
+        )
+    lines.append("")
+
+    # 失败原因分析
+    if stats["failure_categories"]:
+        lines.append("## ❌ 失败原因分析")
+        lines.append("")
+        lines.append("> ⚠️ 基于最近 N 个失败 run 采样分析，用于趋势参考")
+        lines.append("")
+        lines.append("### 按分类统计")
+        lines.append("")
+        total_failures = sum(stats["failure_categories"].values())
+        fc_sorted = sorted(stats["failure_categories"].items(), key=lambda x: -x[1])
+        lines.append("| 分类 | 次数 | 占比 |")
+        lines.append("|------|------|------|")
+        for cat, cnt in fc_sorted:
+            pct = (cnt / total_failures * 100) if total_failures > 0 else 0
+            lines.append(f"| {cat} | {cnt} | {pct:.1f}% |")
+        lines.append("")
+
+        lines.append("### Top 失败 Job")
+        lines.append("")
+        lines.append("| Job 名称 | 失败次数 |")
+        lines.append("|----------|----------|")
+        for job, cnt in stats["failed_jobs_top"].items():
+            lines.append(f"| `{job}` | {cnt} |")
+        lines.append("")
+
+    # Runner 利用率
+    if stats["runner_stats"]:
+        lines.append("## 🏃 Runner 利用率")
+        lines.append("")
+        runner_sorted = sorted(stats["runner_stats"].items(), key=lambda x: -x[1]["jobs"])
+        lines.append("| Runner | Job 数 | 成功 | 失败 | 成功率 | 平均耗时 |")
+        lines.append("|--------|--------|------|------|--------|----------|")
+        for runner, s in runner_sorted:
+            rate = (s["success"] / s["jobs"] * 100) if s["jobs"] > 0 else 0
+            avg = statistics.mean(s["durations"]) if s["durations"] else None
+            lines.append(
+                f"| `{runner}` | {s['jobs']} | {s['success']} | {s['failure']} | {rate:.1f}% | {fmt_duration(avg)} |"
+            )
+        lines.append("")
+
+    # Job 耗时排行
+    if stats["job_time_stats"]:
+        lines.append("## ⏱️ Job 耗时排行 (Top 20 by P95)")
+        lines.append("")
+        job_stats = []
+        for name, durs in stats["job_time_stats"].items():
+            if not durs:
+                continue
+            durs_sorted = sorted(durs)
+            job_stats.append(
+                {
+                    "name": name,
+                    "count": len(durs_sorted),
+                    "avg": statistics.mean(durs_sorted),
+                    "p50": percentile(durs_sorted, 50),
+                    "p95": percentile(durs_sorted, 95),
+                }
+            )
+        job_stats.sort(key=lambda x: -x["p95"])
+        top_n = min(20, len(job_stats))
+        lines.append("| Job 名称 | 次数 | 平均 | P50 | P95 |")
+        lines.append("|----------|------|------|-----|-----|")
+        for j in job_stats[:top_n]:
+            lines.append(
+                f"| `{j['name']}` | {j['count']} | {fmt_duration(j['avg'])} | {fmt_duration(j['p50'])} | {fmt_duration(j['p95'])} |"
+            )
+        lines.append("")
+
+    # 触发事件分布
+    lines.append("## 📋 触发事件分布")
+    lines.append("")
+    evt_sorted = sorted(stats["event_stats"].items(), key=lambda x: -x[1]["total"])
+    lines.append("| 事件类型 | 次数 | 成功 | 失败 | 成功率 |")
+    lines.append("|----------|------|------|------|--------|")
+    for evt, s in evt_sorted:
+        rate = (s["success"] / s["total"] * 100) if s["total"] > 0 else 0
+        lines.append(f"| `{evt}` | {s['total']} | {s['success']} | {s['failure']} | {rate:.1f}% |")
+    lines.append("")
+
+    return "\n".join(lines)
+
+
+# ── 主函数 ───────────────────────────────────────────
+def main():
+    parser = argparse.ArgumentParser(description="CI 可观测性看板 - 生成 Gitea Actions 运行状态报表")
+    parser.add_argument("--days", type=int, default=DEFAULT_DAYS, help=f"统计最近 N 天 (默认 {DEFAULT_DAYS})")
+    parser.add_argument("--output", "-o", type=str, help="输出文件路径 (默认输出到 stdout)")
+    parser.add_argument("--workflow", type=str, help="只统计指定 workflow (如 ci-cd.yml)")
+    parser.add_argument("--gitea-url", type=str, default=os.environ.get("GITEA_URL", DEFAULT_GITEA_URL))
+    parser.add_argument("--repo", type=str, default=os.environ.get("GITEA_REPO", DEFAULT_REPO))
+    parser.add_argument("--token", type=str, default=os.environ.get("GITEA_TOKEN"))
+    parser.add_argument("--username", type=str, default=os.environ.get("GITEA_USERNAME"))
+    parser.add_argument("--password", type=str, default=os.environ.get("GITEA_PASSWORD"))
+    parser.add_argument("--no-job-detail", action="store_true", help="不拉取 job 详情")
+    parser.add_argument("--max-failures", type=int, default=50, help="最多分析多少个失败 run 的 job 详情 (默认 50)")
+    args = parser.parse_args()
+
+    ga = GiteaActions(
+        base_url=args.gitea_url,
+        repo=args.repo,
+        token=args.token,
+        username=args.username,
+        password=args.password,
+    )
+
+    end_date = datetime.now().date()
+    start_date = end_date - timedelta(days=args.days - 1)
+
+    runs = fetch_runs_in_range(ga, start_date, end_date, args.workflow)
+    if not runs:
+        print("[ERROR] 未获取到任何数据", file=sys.stderr)
+        sys.exit(1)
+
+    if not args.no_job_detail:
+        runs = enrich_with_jobs(ga, runs, max_failures=args.max_failures)
+
+    stats = analyze_runs(runs)
+    md = generate_markdown(stats, start_date, end_date, args.repo)
+
+    if args.output:
+        with open(args.output, "w", encoding="utf-8") as f:
+            f.write(md)
+        print(f"[INFO] 报表已保存到 {args.output}", file=sys.stderr)
+    else:
+        print(md)
+
+
+if __name__ == "__main__":
+    main()