Harness Pipeline：给 AI 编程套一条带护栏的跑道

npm install -g @fission-ai/openspec@latest   # 或 brew install openspec
cd your-project && openspec init             # 生成 openspec/ 和各家 agent 的指令

# 然后在你的 AI 工具里走四个命令
# /opsx:explore              还没想清楚时，先当思考搭子
# /opsx:propose add-rate-limit   生成 proposal.md + specs/ + design.md + tasks.md
# /opsx:apply                按 tasks 逐条实现
# /opsx:archive              归档进 specs/ 基线

openspec/
  changes/
    add-rate-limit/
      proposal.md        # 这次改什么、为什么
      design.md          # 关键设计决策
      tasks.md           # 拆到可执行任务
      specs/             # 本次引入的需求与场景
  specs/                 # 已发布的规格基线
  changes/archive/       # 归档后的历史 change

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from app.main import app


@pytest.fixture
def client():
    return TestClient(app)


@pytest.fixture(autouse=True)
def _reset_limiter():
    from app.middleware.rate_limit import limiter
    limiter.reset()
    yield

# features/rate_limit.feature
Feature: API 限流给用户友好的提示

  Scenario: 超过限制时返回友好响应
    Given 客户端 "c3" 已经请求 60 次
    When 客户端 "c3" 再次请求 "/api/v1/search"
    Then 响应状态码是 429
    And 响应不包含 "algorithm" 字段

# tests/test_rate_limit_bdd.py
from pytest_bdd import scenarios, given, when, then, parsers

scenarios("../features/rate_limit.feature")

@given(parsers.parse('客户端 "{cid}" 已经请求 {n:d} 次'))
def _warm_up(client, cid, n):
    for _ in range(n):
        client.get("/api/v1/search", headers={"X-Client-Id": cid})

@when(parsers.parse('客户端 "{cid}" 再次请求 "{path}"'), target_fixture="resp")
def _again(client, cid, path):
    return client.get(path, headers={"X-Client-Id": cid})

@then(parsers.parse("响应状态码是 {code:d}"))
def _status(resp, code):
    assert resp.status_code == code

@then(parsers.parse('响应不包含 "{field}" 字段'))
def _no_field(resp, field):
    assert field not in resp.json()

ruff check app tests          # 风格、import、明显 bug
mypy --strict app             # 类型
bandit -q -r app              # 硬编码密钥、不安全的 yaml.load 之类

<!-- scripts/review-prompt.md -->
你在审一份 AI 生成的 diff。先读规格，再读 diff，只报实质问题。

规格位置：openspec/changes/<change>/specs/

按顺序检查：
1. 实现与 spec 的每条 Requirement / Scenario 是否逐条对应？缺了哪条、哪条被偷换？
2. 有没有"看起来对、其实绕过边界"的地方？重点看 off-by-one、空值、并发、超时。
3. 错误处理是真处理，还是 catch 完就吞？
4. 日志、响应体、异常信息里有没有敏感字段或内部实现细节？
5. 测试测的是行为还是实现（改个变量名就红的那种）？

输出格式，每条一行：
<BLOCKER|MAJOR|MINOR> 文件:行号 — 问题 — 建议改法
没有问题就只输出一行：NO ISSUES FOUND
不要夸奖，不要复述代码，不要给无关建议。

#!/usr/bin/env bash
# scripts/ai_review.sh —— 换个模型审 diff
set -euo pipefail
base="${1:-origin/main}"
git diff "$base...HEAD" > /tmp/review.patch

# 三选一：claude -p / codex exec / opencode run，关键是跟写代码的那个模型换开
{ cat scripts/review-prompt.md; echo; cat /tmp/review.patch; } \
  | claude -p --model "$REVIEW_MODEL" > review-report.md

if grep -q "^BLOCKER" review-report.md; then
  cat review-report.md
  exit 1
fi
echo "AI review passed"

#!/usr/bin/env python3
"""scripts/check_log_privacy.py —— 扫描 logging 调用，禁止敏感字段进日志。"""
import ast
import sys
from pathlib import Path

BANNED = {"phone", "mobile", "email", "token", "password", "id_card", "client_id"}
LOG_FUNCS = {"debug", "info", "warning", "error", "critical", "exception"}


def banned_hits(node: ast.AST) -> set[str]:
    hits: set[str] = set()
    for sub in ast.walk(node):
        if isinstance(sub, ast.Name) and sub.id.lower() in BANNED:
            hits.add(sub.id)
        elif isinstance(sub, ast.Attribute) and sub.attr.lower() in BANNED:
            hits.add(sub.attr)
        elif isinstance(sub, ast.Constant) and isinstance(sub.value, str):
            hits |= {w for w in BANNED if w in sub.value.lower()}
    return hits


def check(path: Path) -> list[str]:
    problems = []
    tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
    for node in ast.walk(tree):
        if not isinstance(node, ast.Call):
            continue
        if not (isinstance(node.func, ast.Attribute) and node.func.attr in LOG_FUNCS):
            continue
        hits: set[str] = set()
        for arg in [*node.args, *(kw.value for kw in node.keywords)]:
            hits |= banned_hits(arg)
        if hits:
            problems.append(f"{path}:{node.lineno}: 日志疑似包含敏感字段 {sorted(hits)}")
    return problems


if __name__ == "__main__":
    root = Path(sys.argv[1] if len(sys.argv) > 1 else "app")
    found = [p for f in root.rglob("*.py") for p in check(f)]
    print("\n".join(found) if found else "log privacy check passed")
    sys.exit(1 if found else 0)

$ python scripts/check_log_privacy.py app
app/middleware/rate_limit.py:5: 日志疑似包含敏感字段 ['client_id']
app/middleware/rate_limit.py:6: 日志疑似包含敏感字段 ['email']

.PHONY: gate-fast gate-pr gate-merge

gate-fast:        # 秒级，pre-commit 里跑
    ruff check app tests
    ruff format --check app tests

gate-pr:          # 分钟级，PR 流水线
    mypy --strict app
    bandit -q -r app
    python scripts/check_log_privacy.py app
    pytest -q tests

gate-merge:       # 合并前，含 BDD 和 AI Review
    pytest -q tests features
    bash scripts/ai_review.sh origin/main

# .github/workflows/harness.yml
name: harness
on: [pull_request]
jobs:
  gates:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with: { fetch-depth: 0 }        # AI Review 要 diff，必须全量 clone
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - run: pip install -r requirements-dev.txt
      - run: make gate-fast
      - run: make gate-pr
      - run: make gate-merge
        env:
          REVIEW_MODEL: ${{ vars.REVIEW_MODEL }}

# .pre-commit-config.yaml —— 本地这一道别省，省了就全堵在 CI 上
repos:
  - repo: local
    hooks:
      - id: gate-fast
        name: gate-fast
        entry: make gate-fast
        language: system
        pass_filenames: false
      - id: log-privacy
        name: log-privacy
        entry: python scripts/check_log_privacy.py app
        language: system
        pass_filenames: false

# openspec/changes/add-rate-limit/proposal.md

## Why
高峰期 /api/v1/search 被少量客户端打爆，影响其他用户。

## What Changes
- 引入 API 网关层的限流中间件（单机滑动窗口）
- 暴露 RateLimitExceeded 领域事件供下游订阅
- 限流触发时不暴露算法细节

## Out of Scope
- 分布式限流（下一个 change）
- 配额管理 UI

## Impact
- Affected spec: api-gateway（新增 rate-limit 能力）
- Affected code: app/middleware/, app/events/

# openspec/changes/add-rate-limit/specs/api-gateway/spec.md

## ADDED Requirements

### Requirement: 单客户端限流
系统 SHALL 限制单个客户端在 60s 内对 /api/v1/search 的请求次数。

#### Scenario: 在限制内
- **WHEN** 客户端 60s 内请求不超过 60 次
- **THEN** 全部正常处理（200）

#### Scenario: 超出限制
- **WHEN** 客户端 60s 内请求超过 60 次
- **THEN** 返回 429
- **AND** 响应体为 {"error": "rate_limited", "retry_after": <秒>}
- **AND** 响应体 MUST NOT 包含算法、窗口大小等内部细节
- **AND** 触发 RateLimitExceeded 事件

### Requirement: 匿名客户端
未携带 X-Client-Id 的请求 SHALL 按匿名身份限流，限制为 10 req/min。

### Requirement: 健康检查豁免
/health 路径 SHALL NOT 计入任何限流计数。

# tests/test_rate_limit.py（client / _reset_limiter 来自 conftest.py）


def test_within_limit_returns_200(client):
    for _ in range(60):
        r = client.get("/api/v1/search", headers={"X-Client-Id": "c1"})
        assert r.status_code == 200


def test_over_limit_returns_429_with_retry_after(client):
    headers = {"X-Client-Id": "c2"}
    for _ in range(60):
        client.get("/api/v1/search", headers=headers)
    r = client.get("/api/v1/search", headers=headers)
    assert r.status_code == 429
    body = r.json()
    assert body["error"] == "rate_limited"
    assert body["retry_after"] > 0
    assert "window" not in body and "algorithm" not in body   # 不泄露内部细节


def test_health_check_not_counted(client):
    for _ in range(100):
        assert client.get("/health").status_code == 200

make gate-fast && make gate-pr && make gate-merge

# app/middleware/rate_limit.py（节选）
from prometheus_client import Counter

blocked_total = Counter(
    "api_rate_limit_blocked_total",
    "Requests blocked by rate limiter",
    ["endpoint", "client_tier"],
)


async def rate_limit_middleware(request, call_next):
    if is_over_limit(request):
        blocked_total.labels(
            endpoint=request.url.path,
            client_tier=classify(request),
        ).inc()
        return JSONResponse(
            status_code=429,
            content={"error": "rate_limited", "retry_after": 60},
        )
    return await call_next(request)

# .importlinter
[importlinter]
root_package = app

[importlinter:contract:layers]
name = 分层依赖：上层可依赖下层，反过来不行
type = layers
layers =
    app.api
    app.service
    app.repository
    app.types

@startmindmap
<style>
mindmapDiagram {
  node {
    BackgroundColor #F8F9FA
    RoundCorner 10
    Padding 10
    FontSize 13
  }
  :depth(0) {
    BackgroundColor #1E3A5F
    FontColor white
    FontSize 18
    FontStyle bold
  }
  :depth(1) {
    FontSize 15
    FontStyle bold
  }
}
</style>

* Harness Pipeline
** 旧流水线漏掉的五件事
*** 意图跑偏
*** 契约被绕过
*** 体验不对路
*** 本地知识被无视
*** 假设落空
** 四层骨架
*** SDD\nOpenSpec + DDD 三件套
*** TDD\n人写 3-5 条核心契约
*** BDD\npytest-bdd 跑场景
*** MDD\n假设→指标→阈值
** 三道闸门
*** 静态分析\nruff/mypy/bandit
*** AI Review\n换模型 + prompt 原文
*** 规则检查\n日志隐私脚本
** 接进 CI
*** Makefile 三个 target
*** pre-commit + Actions
*** 时间预算，超了就会被绕过
** 业界趟平的路
*** 变异测试\n替代覆盖率
*** 属性测试\nHypothesis
*** 契约测试\nSchemathesis/Pact
*** 架构适应度函数\nimport-linter
*** 护具自己也要审\nQuartetFuzz 四原则
** 十个陷阱
*** 自己审自己
*** 门禁可绕 / 太慢
*** flaky 靠重跑
*** 覆盖率当门禁
*** spec 被顺手改
*** 三千行大 MR
** 不走全程的情况
*** 文案样式：只跑单测
*** 钱/权限/数据：全程 + 人拍板
*** 最小组合：规则+契约+门禁
@endmindmap