scaffold/template/.claude/rules/backend.md

paths

paths
backend/**

backend

FastAPI 服务（port 8000）的技术栈、目录、分层、路由、中间件与配置规范。

概览

职责：WebSocket 代理层 + 文件服务 + 会话持久化。 Backend 不连接任何 LLM——所有大模型交互都在 Agent。Backend 只做 WebSocket 代理 + 数据持久化。

技术栈

库	用途
FastAPI + uvicorn	Web 框架 + ASGI 服务器
SQLAlchemy 2.x (async)	ORM
SQLite	轻量数据库（单文件，无需独立服务）
Alembic	数据库迁移
loguru	结构化日志
pydantic-settings	环境变量管理
websockets	Agent WebSocket 客户端
pytest + httpx + pytest-asyncio	测试

包管理（uv）

uv add fastapi uvicorn "sqlalchemy[asyncio]" aiosqlite alembic loguru pydantic-settings websockets pytest httpx pytest-asyncio
uv run python -m src.main          # 运行服务
uv run pytest                      # 运行测试

关键文件：pyproject.toml（依赖声明）、uv.lock（锁文件，必须提交 git）、alembic.ini（迁移配置）

目录结构

backend/
├── alembic.ini                  # Alembic 入口配置
├── alembic/
│   ├── env.py                   # 迁移环境（导入所有 Model）
│   └── versions/                # 自动生成的迁移脚本
├── src/
│   ├── main.py                  # FastAPI app 实例、启动/关闭事件
│   ├── domain/                  # 领域层，纯逻辑，零框架/IO 依赖
│   │   ├── entities.py          # Session 聚合根（纯 Python dataclass，含状态机方法）
│   │   ├── value_objects.py     # SessionStatus（StrEnum，含 can_recover 规则）
│   │   └── ports.py             # SessionRepository Protocol（仓储端口）
│   ├── routers/
│   │   ├── ws.py                # WebSocket 代理 /ws/{session_id}
│   │   └── files.py             # 文件下载 + 预览路由
│   ├── services/
│   │   ├── agent_client.py      # 连接 Agent WebSocket
│   │   └── session_service.py   # 会话用例编排（依赖 SessionRepository 端口）
│   ├── schemas/
│   │   ├── messages.py          # Pydantic WebSocket 消息模型
│   │   └── session.py           # Pydantic 会话读写 schema（与 ORM 解耦）
│   ├── middleware/
│   │   ├── request_id.py        # 每个请求注入唯一 X-Request-Id
│   │   ├── access_log.py        # 请求/响应结构化日志
│   │   └── error_handler.py     # 全局未捕获异常 → 标准错误响应
│   ├── exceptions/
│   │   ├── base.py              # AppException 基类
│   │   └── handlers.py          # FastAPI exception_handler 注册
│   ├── db/
│   │   ├── base.py              # async_engine、AsyncSessionLocal、Base 声明基类
│   │   ├── models.py            # SessionRecord ORM 映射（纯持久化，无业务逻辑）
│   │   └── repositories.py      # SQLAlchemySessionRepository（实现 SessionRepository 端口）
│   ├── utils/                   # 横切纯函数工具（无业务逻辑、无状态）
│   │   ├── ids.py               # request_id / uuid 生成
│   │   └── timing.py            # 计时、超时上下文管理器
│   └── core/
│       ├── config.py            # Settings（pydantic-settings）
│       ├── deps.py              # FastAPI 依赖注入（get_db、get_session_repo、get_settings）
│       └── logging.py           # loguru 初始化、sink 配置
└── tests/
    ├── conftest.py              # 测试 DB（in-memory SQLite）、app fixture
    ├── test_session_entity.py   # Session 领域实体单元测试（classicist，不 mock）
    ├── test_ws.py
    ├── test_files.py
    └── test_session_service.py

分层映射（六边形架构）

backend 是支撑域，遵循六边形架构，依赖恒向内：

层	目录	职责
接口适配器	routers/	WS/HTTP 入站适配，仅转译协议
应用（用例）	services/	会话用例编排，依赖 SessionRepository 端口
领域	domain/	Session 聚合根、SessionStatus 值对象、SessionRepository 端口；纯逻辑零 IO
基础设施	db/repositories.py、agent_client	实现 SessionRepository 端口、出站适配

路由层规范

• 路由函数只做：参数校验 + 调用 service + 返回响应
• 禁止在路由函数中写业务 if/else
• 所有 DB 操作必须通过 session_service，禁止在路由层直接操作 DB
• REST 路由统一挂 /api/v1 前缀（下载/预览等）；WS 消息带 v 字段（见根「协议版本化」）
• 会话记录保留 DB_RETENTION_DAYS（默认 30 天），lifespan 定时任务软删→硬删过期记录

RESTful 资源命名规范

项	规范
资源	名词复数：/api/v1/sessions、/api/v1/sessions/{session_id}/export
路径段	小写 + kebab-case，不放动词（动作用 HTTP method）
方法	GET 读 / POST 建 / PUT 全量改 / PATCH 局部改 / DELETE 删
JSON 字段	snake_case
版本	/api/v1 前缀
状态码	语义化 200/201/204/4xx/5xx

例：GET /api/v1/sessions/{session_id}/export。路径/字段命名靠 review + 可选 spectral（OpenAPI lint）；HTTP 语义靠 review。

Swagger / OpenAPI 文档

• FastAPI 自动生成 /docs（Swagger UI）、/redoc、/openapi.json
• 由 .env 的 ENABLE_API_DOCS 控制：dev 默认开，prod 关（docs_url=None 等）
• Settings 字段 enable_api_docs: bool = True，映射环境变量 ENABLE_API_DOCS

中间件（middleware/）

request_id.py

每个请求（含 WebSocket 握手）注入 X-Request-Id，传入 loguru 上下文，在所有日志条目中携带，便于全链路追踪。

access_log.py

记录每个 HTTP 请求的结构化日志（method、path、status_code、duration_ms、request_id）。WebSocket 连接记录 connect / disconnect 事件。

error_handler.py

捕获所有未处理异常，返回统一错误结构：

{
  "error": {
    "code": "INTERNAL_ERROR",
    "message": "服务器内部错误",
    "request_id": "xxx"
  }
}

生产环境不暴露堆栈信息；开发环境在日志中完整记录。

异常体系（exceptions/）：业务错误必须抛 AppException 子类（携带 code/message/status_code），禁止在路由层直接 raise HTTPException。在 main.py 注册 app.add_exception_handler(AppException, app_exception_handler) 与 app.add_exception_handler(Exception, global_exception_handler)。

配置（core/config.py）

配置项见 .env.example（提交 git 的模板）；真实配置放 .env（gitignore）。.env 用 SCREAMING_SNAKE，Settings 字段用 snake_case；env 映射（agent_ws_url ↔ AGENT_WS_URL）为 pydantic-settings 默认行为，无需 alias_generator。

class Settings(BaseSettings):
    env: Literal["dev", "prod"] = "dev"
    agent_ws_url: str = "ws://localhost:8001"
    work_dir: Path = Path.home() / "app-data"
    database_url: str = f"sqlite+aiosqlite:///{Path.home()}/app-data/app.db"
    port: int = 8000
    log_level: str = "DEBUG"
    enable_api_docs: bool = True

    model_config = SettingsConfigDict(
        env_file=".env", env_file_encoding="utf-8",
    )

• 缺失/类型错误在启动时即报错（fail-fast）
• 优先级：环境变量 > .env > 默认值
• 敏感配置只从环境读取，.env 不提交 git

WebSocket 代理核心

重连入口（routers/ws.py）：

@router.websocket("/ws/{session_id}")
async def ws_proxy(ws: WebSocket, session_id: str, repo: SessionRepository = Depends(get_session_repo)):
    await ws.accept()
    checkpoint = await session_service.load_checkpoint(repo, session_id)
    # 将 checkpoint 信息附在 start_session 消息中转发给 Agent
    async with agent_client.connect(session_id, checkpoint=checkpoint) as agent_ws:
        await asyncio.gather(forward_to_agent(ws, agent_ws), forward_to_browser(agent_ws, ws))

详见 backend-concurrency.md（双向转发 + Task 管理）与 backend-db.md（checkpoint）。

日志规范（core/logging.py）

loguru，两个 sink：stderr（dev 人类可读彩色）+ {WORK_DIR}/logs/backend.log（JSON 每行一条，所有环境）。文件 sink 配 rotation="100 MB"、retention="14 days"、compression="zip"。级别 DEBUG(dev)/INFO(prod)。每条日志必须携带 request_id（loguru contextualize 注入）。禁止 print() 与 stdlib logging，统一 from loguru import logger。

测试规范

• 领域实体（classicist）：test_session_entity.py 直接实例化 Session，验证状态机方法与不变量，零 mock，零 IO
• service 层（mockist）：test_session_service.py mock SessionRepository 端口，验证用例编排（状态流转调用顺序）
• 集成（真实 DB）：conftest.py 提供 in-memory SQLite + SQLAlchemySessionRepository；test_ws.py mock Agent WS，验证消息原样转发
• checkpoint 测试：验证不同 status 下重连时返回正确的 Session 数据，can_recover() 结果符合预期
• 每次 CI 前自动运行 alembic upgrade head（针对测试 DB）