fix(ai): 修 ruff lint 三处错误——CI 由红转绿

`4762358` 与 `86d63ab` 两次 CI 均因 `ruff check` 失败而红。

**根因是我的门禁自查方法错了**：用 `ruff check . | tail -1; echo $?` 读到的是
`tail` 的退出码而不是 ruff 的，于是把一次真实的 lint 失败读成了通过——先前声称
的「四道门禁全绿」是基于这个坏掉的检查得出的。已在 ci.yml 的 Gates 步骤就地注明
此坑，并改为逐条分开跑读各自退出码。

三处都是实际问题，非误报：
- `client.py:138` PYI034：`__enter__` 的返回注解硬编码为类名，应为 `Self`
  （子类化后注解会失准）
- `test_video.py:58` C408：`dict()` 调用改字面量
- `test_layering.py:85` PIE810：两次 `endswith` 合并为一次元组调用

修复后在与 CI 同构的隔离 venv（同镜像源、同 requirements-dev.txt，实测 ruff
版本与本地一致 0.16.8）逐道复验：secret-scan / format / lint / pytest 四道
退出码均为 0，100 测试全过。

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

.env.sample

@@ -0,0 +1,32 @@
+# patbond-ai 配置样例。复制为 .env 后填入实际值——.env 已被 gitignore。
+#
+# 凭证纪律（ADR-021）：真实密钥只允许存在于被忽略的文件中，本样例只放占位值。
+# 被拦下时的第一动作是去服务商控制台轮换/禁用该密钥，然后才是清理提交。
+
+# ---- 火山方舟（Ark）----
+# 密钥无默认值。缺失时该适配器不装配，服务仍可启动并提供健康检查。
+ARK_API_KEY=your-ark-api-key
+# 北京区官方地址，一般无需改。
+# ARK_BASE_URL=https://ark.cn-beijing.volces.com/api/v3
+# 单次 HTTP 往返超时（秒）。**不是生成耗时**——提交接口是异步的（只入队即返回），
+# 调大成「等出图」会重新引入长连接占用与计费歧义。
+# ARK_TIMEOUT_SEC=10
+# 视频模型 ID，仅作兜底。正常路径由 Java 侧传入任务行冻结的模型快照——
+# 适配层不自行选模型，否则任务记录与实际调用会不一致、事后无法对账。
+ARK_VIDEO_MODEL=
+# 厂商 execution_expires_after（秒），区间 [3600, 259200]。
+# 默认取下限 1 小时：厂商默认 48 小时对交互式应用过长，被放弃的任务会挂两天。
+# ARK_TASK_EXPIRY_SEC=3600
+
+# ---- 服务身份 JWT（ADR-023 §5）----
+# 与 patbond-creation 持有的私钥配对的 RS256 公钥。注意这是一对**独立于用户
+# token 的**密钥：现有 JwtVerifier 只验签名、不校验 aud/iss，复用同一密钥对会
+# 让服务票被四个 Java 模块当成合法用户 token 接受。
+PATBOND_SERVICE_JWT_PUBLIC_KEY=
+# PATBOND_SERVICE_JWT_ISSUER=patbond-api
+# audience 必须强制校验，否则独立密钥对失去意义。
+# PATBOND_SERVICE_JWT_AUDIENCE=patbond-ai
+# PATBOND_SERVICE_JWT_TYPE=service
+
+# ---- 其他 ----
+# PATBOND_AI_LOG_LEVEL=INFO

.gitea/workflows/ci.yml

@@ -0,0 +1,65 @@
+# Gitea Actions 门禁：与本地门禁同一套命令（规范见 patbond-doc
+# docs/development/git-workflow.md）。零外部 action / 零 GitHub 依赖，
+# 原因见 patbond-api 的 ci.yml 头注释。
+# pip 与 PyPI 走腾讯云镜像，与 patbond-doc 的 mkdocs 安装同一策略。
+#
+# 本仓是唯一存放 AI 服务商凭证的仓库，因此 Secret scan 是第一个 step，
+# 且先于任何依赖安装执行（ADR-021、ADR-023 §12）。
+name: CI
+
+on:
+  push:
+    branches: [dev]
+  pull_request:
+
+concurrency:
+  group: ci-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  ai-gates:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout (manual)
+        run: |
+          git init -q .
+          AUTH_URL=$(echo "${{ github.server_url }}" | sed "s#https://#https://oauth2:${{ github.token }}@#")
+          git remote add origin "$AUTH_URL/${{ github.repository }}.git"
+          git fetch -q --depth 1 origin "+${{ github.ref }}:refs/ci-head"
+          git checkout -q refs/ci-head
+      # 凭证防泄漏兜底（ADR-021）：与本地 pre-commit 同一脚本、同一规则表，
+      # 四仓副本内容同构，扫全部已跟踪文件（本次 push 变更的超集）。
+      - name: Secret scan
+        run: sh scripts/check-secrets.sh --all
+      - name: Set up Python
+        run: |
+          apt-get update -qq
+          apt-get install -y -qq --no-install-recommends python3 python3-venv python3-pip
+          python3 -m venv .venv
+          echo "$PWD/.venv/bin" >> "$GITHUB_PATH"
+          echo "PIP_INDEX_URL=https://mirrors.cloud.tencent.com/pypi/simple" >> "$GITHUB_ENV"
+      - name: Install dependencies
+        run: |
+          python -m pip install -q --upgrade pip
+          # requirements-dev.txt 落地前，门禁只装 lint/test 工具本身，
+          # 使 CI 从建仓第一天就是绿的而非「等代码来了再配」。
+          if [ -f requirements-dev.txt ]; then
+            pip install -q -r requirements-dev.txt
+          else
+            pip install -q ruff pytest
+          fi
+          if [ -f requirements.txt ]; then pip install -q -r requirements.txt; fi
+      # 与本地门禁同一套命令。ruff 兼任 format 检查与 lint，
+      # 对应 patbond-flutter 的 dart format + flutter analyze 两步。
+      #
+      # 本地自查时注意：`ruff check . | tail -1; echo $?` 读到的是 **tail** 的
+      # 退出码，不是 ruff 的——这样会把一次真实的 lint 失败读成通过（本仓已因此
+      # 红过一次）。逐条分开跑，或用 ${PIPESTATUS[0]}。
+      - name: Gates
+        run: |
+          python --version
+          ruff format --check .
+          ruff check .
+          # 尚无测试文件时 pytest 退出码为 5（no tests collected），
+          # 该情形视为通过；真实失败仍会以非 0/5 退出。
+          pytest -q || [ $? -eq 5 ]

.gitignore

@@ -0,0 +1,57 @@
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# Keep application sources visible even if a global gitignore ignores lib/.
+!/lib/
+!/lib/**
+
+# Virtualenvs
+.venv/
+venv/
+env/
+ENV/
+
+# Tooling caches
+.pytest_cache/
+.ruff_cache/
+.mypy_cache/
+.coverage
+.coverage.*
+htmlcov/
+coverage.xml
+
+# 凭证载体一律不入库（ADR-021）。真实值只存在于被忽略的文件中，
+# 占位样例走 .env.sample / *.example（check-secrets.sh 的 DENY_NAME_OK）。
+.env
+.env.*
+!.env.sample
+!.env.example
+credentials*
+*.pem
+*.key
+
+# 本地产物与临时文件
+*.log
+*.swp
+.DS_Store
+tmp/
+.idea/
+.vscode/

README.md

@@ -0,0 +1,128 @@
+# patbond-ai
+
+Patbond 的 AI 生成服务（Python）。负责调用 AI 服务商 API 完成图片与视频生成，
+是四仓中唯一直接持有 AI 服务商凭证的仓库。
+
+架构与边界的完整决策见 **patbond-doc `docs/architecture/decisions.md` ADR-023**。
+以下为要点，改动前请以 ADR 原文为准。
+
+## 职责边界
+
+- **只负责调用 AI 服务商 API。** 不访问数据库——生成任务状态的唯一事实来源是
+  Java 侧的 `creation.generation_jobs`。
+- **不持有对象存储凭证。** 产物由 Java 下发预签名 PUT 地址，本服务直传后回报，
+  再由 Java 核实并建 `media.assets` 行（即客户端两步上传，本服务站在客户端位置）。
+- **对上游同步或异步一律对外呈现为异步**：`submit` 立即返回任务 id，同步型
+  服务商（长连接等待返回）在本服务内部以后台任务承载。这样 Java 侧无长连接占线程，
+  且 `provider_request_id` 立即可得，超时不再产生「是否已扣费」的歧义。
+
+## 与 Java 侧的接口
+
+- 传输为 **REST（JSON）**，非 gRPC（理由见 ADR-023 §4：避免引入第二套无门禁契约体系）。
+- 契约为本仓内独立的 **`openapi-internal.yaml`**，独立版本号，
+  **不进** patbond-api 的四模块快照锁，**也不写入**面向客户端的 `openapi.yaml`。
+- 鉴权为 **服务身份 JWT**（独立密钥对，非用户 token 转发）：本服务强制校验
+  `aud=patbond-ai` 与 `typ=service`。
+- Java 主动轮询本服务查询进度；回调是将来的可选优化，当前不启用。
+- 请求体**不含 `userId`**（数据最小化）：完成工作只需 `jobId`、生成参数与预签名上传地址。
+
+## 代码结构
+
+分四层，依赖**单向**向内（`config` → `providers` → `domain`），由
+`tests/test_layering.py` 自动守着——靠纪律守不住的东西，靠测试才守得住。
+
+```
+app/
+├── domain/generation.py      与服务商无关的领域模型
+├── providers/
+│   ├── base.py               生成能力抽象（新增供应商 = 实现它）
+│   ├── registry.py           provider_code → 适配器
+│   └── volcengine/
+│       ├── client.py         HTTP 传输与鉴权（不认识「视频」）
+│       ├── errors.py         火山错误码 → 领域 Failure
+│       ├── models.py         厂商请求/响应 DTO
+│       └── video.py          Seedance 视频业务
+├── config.py                 集中配置（全部旋钮从环境变量读）
+└── assembly.py               装配根（唯一同时知道配置与具体供应商的地方）
+```
+
+**为什么这么切**：
+
+- **`domain/` 对厂商与 httpx 零依赖。** 这是「换供应商不动上层」的支点。判断一个
+  概念该放哪：问「换一家供应商，它还在吗」——`Retry`/`UserFacing` 描述的是
+  Worker 与客户端该怎么做，与谁家无关，故在领域层；`expired` 状态、
+  `QuotaExceeded` 这类码是火山特有的，故在适配层。
+- **`client.py` 与 `video.py` 分开。** client 只管鉴权、超时、错误解包，不认识
+  「视频」。以后接火山的**图片或语音**接口时新建兄弟模块复用同一个 client，
+  而不是再抄一遍鉴权与错误解包——这是拆它出来的全部理由。
+- **适配层不读配置**，全部通过构造参数注入。否则单测得摆弄环境变量，且同一进程
+  内没法跑两套不同配置的实例。
+
+### 新增一家供应商 / 一种能力
+
+1. 实现 `app/providers/base.py` 的 `GenerationProvider`；
+2. 在 `app/assembly.py` 的 `build_registry()` 里登记一行；
+3. 目录表的 `provider_code` 与适配器的 `provider_code` 必须同值——Java 侧按
+   任务行冻结的 `provider_code_snapshot` 查适配器（**不按当前目录**，因为目录
+   可能被改指向，而在途任务必须仍由当初提交它的适配器轮询）。
+
+上层与领域层**一行都不用改**。
+
+## 已接入的服务商
+
+| 服务商 | 能力 | provider_code | 状态 |
+| --- | --- | --- | --- |
+| 火山方舟 Seedance | 视频生成 | `volcengine-seedance` | 提交与查询已实现，产物转存待存储决策 |
+
+接口事实与错误码映射的完整取证见 patbond-doc
+`docs/development/providers/volcengine-seedance.md`。其中两条尤其要记住：
+
+- **火山不支持幂等键**，「提交超时后不知道有没有扣费」只能由 `max_attempts` 兜住；
+- **产物 URL 只有 24 小时有效期**，因此厂商 URL 不能当存储用，必须在时限内转存。
+
+## 配置
+
+全部旋钮见 `app/config.py`，常用项：
+
+| 环境变量 | 默认值 | 说明 |
+| --- | --- | --- |
+| `ARK_API_KEY` | _（无）_ | 火山方舟密钥。**缺失时该适配器不装配**，服务仍可启动 |
+| `ARK_BASE_URL` | 北京区官方地址 | |
+| `ARK_TIMEOUT_SEC` | `10` | **单次 HTTP 往返**超时，不是生成耗时 |
+| `ARK_VIDEO_MODEL` | _（无）_ | 仅作兜底；正常由 Java 侧传入任务行的模型快照 |
+| `ARK_TASK_EXPIRY_SEC` | `3600` | 厂商 `execution_expires_after`，区间 `[3600, 259200]`。默认取下限——厂商默认 48 小时对交互式应用过长 |
+| `PATBOND_SERVICE_JWT_PUBLIC_KEY` | _（无）_ | 校验 Java 侧服务票的 RS256 公钥 |
+| `PATBOND_SERVICE_JWT_AUDIENCE` | `patbond-ai` | **必须强制校验**，否则独立密钥对失去意义 |
+
+非法值**响亮失败而不静默回落**：一次配置笔误变成难查的行为偏差是最糟的结果。
+
+## 本地门禁
+
+与 CI 同一套命令（`.gitea/workflows/ci.yml`）：
+
+```bash
+sh scripts/check-secrets.sh --all   # 凭证防泄漏全量自查
+ruff format --check .
+ruff check .
+pytest -q
+```
+
+**每人每仓启用一次** pre-commit 防泄漏钩子：
+
+```bash
+git config core.hooksPath scripts/hooks
+```
+
+`scripts/check-secrets.sh` 是四仓同构副本，**调整规则时四仓同步提交**。
+它覆盖 AI 服务商密钥形态（`sk-ant-`、`sk-`/`sk-proj-`/`sk-svcacct-`）与
+`api_key`/`auth_token` 实值赋值——本仓正是这些规则的主要适用对象。
+
+## 凭证纪律
+
+真实密钥**只允许存在于被 gitignore 的文件中**（`.env`），占位样例进 `.env.sample`。
+拦下真实凭证时的第一动作是**去服务商控制台轮换/禁用该密钥**，然后才是清理提交。
+
+## 分支与发布
+
+`dev` 直推 + CI 绿，`main` 经 PR 合并。本仓纳入四仓 tag 同步与发布回归
+（ADR-023 §12）。发布流程见 patbond-doc `docs/development/releases.md`。

app/assembly.py

@@ -0,0 +1,53 @@
+"""装配根：把配置变成可用的适配器。
+
+**这是唯一允许同时知道「配置」与「具体供应商」的地方。** 其余模块只依赖抽象：
+适配器通过构造参数拿配置（而非自己读环境变量），上层通过 `ProviderRegistry`
+按 `provider_code` 拿适配器。新增一家供应商只需在此登记一行。
+
+`app/providers/base.py`、`registry.py` 与 `app/domain/` 对具体供应商零依赖，
+这一点由 `tests/test_layering.py` 自动守着——靠纪律守不住，靠测试才守得住。
+"""
+
+from __future__ import annotations
+
+import logging
+
+from app.config import Settings
+from app.providers.base import GenerationProvider
+from app.providers.registry import ProviderRegistry
+from app.providers.volcengine.client import ArkClient
+from app.providers.volcengine.video import SeedanceVideoProvider
+
+log = logging.getLogger(__name__)
+
+
+def build_registry(settings: Settings) -> ProviderRegistry:
+    """按配置装配所有可用的适配器。
+
+    **缺凭证的供应商被跳过而不是让装配失败**：本地只配了一家凭证时，服务仍应能
+    起来并提供其余能力与健康检查。真正的失败点在调用时——那时 `find()` 返回
+    None，调用方按「适配器不可用」处理，而不是把一个半死的实例放出去。
+    """
+    providers: list[GenerationProvider] = []
+
+    volcengine = settings.volcengine
+    if volcengine.api_key:
+        providers.append(
+            SeedanceVideoProvider(
+                ArkClient(
+                    api_key=volcengine.api_key,
+                    base_url=volcengine.base_url,
+                    timeout_sec=volcengine.timeout_sec,
+                ),
+                default_model=volcengine.video_model,
+                task_expiry_sec=volcengine.task_expiry_sec,
+            )
+        )
+    else:
+        # 警告而非静默：这是最容易被忽略的配置缺失，而症状（任务永远排队）
+        # 与它相隔很远。
+        log.warning("ARK_API_KEY 未配置，火山方舟视频适配器未装配")
+
+    registry = ProviderRegistry(providers)
+    log.info("已装配 %d 个生成适配器：%s", len(registry), sorted(registry.codes))
+    return registry

app/config.py

@@ -0,0 +1,125 @@
+"""集中配置：全部旋钮从环境变量读，一处定义。
+
+为什么不用 pydantic-settings：当前运行时依赖只有 httpx 一个，为十来个环境变量
+引入一个框架不划算。需要嵌套配置或复杂校验时再换。
+
+**凭证纪律**（ADR-021）：密钥只从环境变量读，**绝不写默认值**。缺失时
+`from_env()` 不抛异常——服务仍可启动并提供健康检查，只在真正要调用厂商时才
+响亮失败。这与 Java 侧 `JwtVerifier`/`ServiceTokenSigner` 的既有约定一致：
+缺配置是服务端错误，不该伪装成上游故障，但也不该让整个进程起不来。
+"""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass, field
+
+from app.providers.volcengine.models import DEFAULT_BASE_URL
+
+
+def _env(name: str, default: str = "") -> str:
+    return os.environ.get(name, default).strip()
+
+
+def _env_int(name: str, default: int) -> int:
+    raw = _env(name)
+    if not raw:
+        return default
+    try:
+        return int(raw)
+    except ValueError as exc:
+        # 静默回落到默认值会让一次配置笔误变成难查的行为偏差。
+        raise ValueError(f"环境变量 {name} 不是合法整数：{raw!r}") from exc
+
+
+def _env_float(name: str, default: float) -> float:
+    raw = _env(name)
+    if not raw:
+        return default
+    try:
+        return float(raw)
+    except ValueError as exc:
+        raise ValueError(f"环境变量 {name} 不是合法数字：{raw!r}") from exc
+
+
+@dataclass(frozen=True)
+class VolcengineSettings:
+    """火山方舟（Ark）接入配置。
+
+    :param api_key: `ARK_API_KEY`。**无默认值**，缺失时适配层在调用点报错。
+    :param timeout_sec: 单次 HTTP 往返超时。**不是生成耗时**——提交与查询都是
+        短请求（提交只入队即返回），把它调大成「等出图」会重新引入长连接占用与
+        计费歧义，见 ADR-023 §4。
+    :param video_model: 视频生成的默认模型 ID。仅作**兜底**：正常路径由 Java 侧
+        从目录表的快照列传入，适配层不自行选模型（选型是目录的职责，且任务行里
+        冻结的快照必须与实际调用的模型一致）。
+    :param task_expiry_sec: 传给厂商的 `execution_expires_after`。厂商取值范围
+        `[3600, 259200]`，默认 48 小时对交互式应用过长，故默认取下限 1 小时。
+    """
+
+    api_key: str = ""
+    base_url: str = DEFAULT_BASE_URL
+    timeout_sec: float = 10.0
+    video_model: str = ""
+    task_expiry_sec: int = 3600
+
+    @classmethod
+    def from_env(cls) -> VolcengineSettings:
+        return cls(
+            api_key=_env("ARK_API_KEY"),
+            base_url=_env("ARK_BASE_URL", cls.base_url),
+            timeout_sec=_env_float("ARK_TIMEOUT_SEC", cls.timeout_sec),
+            video_model=_env("ARK_VIDEO_MODEL"),
+            task_expiry_sec=_env_int("ARK_TASK_EXPIRY_SEC", cls.task_expiry_sec),
+        )
+
+    def __post_init__(self) -> None:
+        # 厂商硬性区间。设成区间外的值会在提交时被厂商拒（InvalidParameter），
+        # 那要花一次往返才知道，不如在启动时就说清楚。
+        if not 3600 <= self.task_expiry_sec <= 259200:
+            raise ValueError(
+                f"ARK_TASK_EXPIRY_SEC 必须在 [3600, 259200] 之间（厂商限制），"
+                f"当前 {self.task_expiry_sec}"
+            )
+
+
+@dataclass(frozen=True)
+class ServiceAuthSettings:
+    """校验 Java 侧服务身份 JWT 所需的配置（ADR-023 §5）。
+
+    :param public_key: 与 `patbond-creation` 持有的私钥配对的 RS256 公钥。
+        注意这是一对**独立于用户 token 的**密钥——理由见 ADR-023 §5。
+    :param audience: 必须强制校验。缺了这一步，任何用同一密钥签出的 token 都能
+        进来，而独立密钥对的意义正在于此。
+    """
+
+    public_key: str = ""
+    issuer: str = "patbond-api"
+    audience: str = "patbond-ai"
+    required_type: str = "service"
+
+    @classmethod
+    def from_env(cls) -> ServiceAuthSettings:
+        return cls(
+            public_key=_env("PATBOND_SERVICE_JWT_PUBLIC_KEY"),
+            issuer=_env("PATBOND_SERVICE_JWT_ISSUER", cls.issuer),
+            audience=_env("PATBOND_SERVICE_JWT_AUDIENCE", cls.audience),
+            required_type=_env("PATBOND_SERVICE_JWT_TYPE", cls.required_type),
+        )
+
+
+@dataclass(frozen=True)
+class Settings:
+    """本服务的全部配置。"""
+
+    volcengine: VolcengineSettings = field(default_factory=VolcengineSettings)
+    service_auth: ServiceAuthSettings = field(default_factory=ServiceAuthSettings)
+    log_level: str = "INFO"
+
+    @classmethod
+    def from_env(cls) -> Settings:
+        return cls(
+            volcengine=VolcengineSettings.from_env(),
+            service_auth=ServiceAuthSettings.from_env(),
+            log_level=_env("PATBOND_AI_LOG_LEVEL", cls.log_level).upper(),
+        )

app/domain/generation.py

@@ -0,0 +1,156 @@
+"""与服务商无关的生成领域模型。
+
+**本模块不得 import 任何服务商代码，也不得 import httpx。** 这不是洁癖，是本层
+存在的唯一理由——Java 侧、未来的 gpt-image-2、以及任何第三家供应商都对着这一层
+编程。一旦这里漏进一个厂商概念，「换供应商不动上层」就不再成立，而那正是分层
+想买到的东西。
+
+判断某个概念该放这里还是放 `providers/<厂商>/`：问「换一家供应商，这个概念还在
+吗」。`Retry`/`UserFacing` 描述的是 **Worker 与客户端该怎么做**，与谁家供应商
+无关，故在此；`expired` 状态、`QuotaExceeded` 这类码是火山特有的，故不在此。
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+
+
+class MediaKind(Enum):
+    """生成产物的类型，与 `creation.generation_jobs.media_kind` 同值。"""
+
+    IMAGE = "image"
+    VIDEO = "video"
+
+
+class Retry(Enum):
+    """失败后该怎么办。由服务商适配层判定，由 Java 侧的 Worker 执行。"""
+
+    #: 不可重试：同样的输入必然得到同样的结果（参数错、内容被拒、鉴权失败）。
+    NEVER = "never"
+    #: 可重试且应尽快：限流、突发流量保护、服务过载。退避几秒到几十秒。
+    BACKOFF = "backoff"
+    #: 可重试但要等很久：额度/配额耗尽，通常要等到下一个计费周期或充值之后。
+    #: Worker 不该把重试预算浪费在这上面，应直接终态失败并让用户看到出口。
+    QUOTA = "quota"
+
+
+class UserFacing(Enum):
+    """给终端用户的呈现类别，对应客户端的两个 429 分支（ADR-023 §9）。
+
+    「配额耗尽」与「频率过快」必须分开：前者等的是明天或充值（弹 sheet 给出口），
+    后者等的是几秒（弹 SnackBar 可自动重试）。合成一类必然有一半场景的文案和
+    出口是错的。
+    """
+
+    QUOTA_EXHAUSTED = "quota_exhausted"
+    RATE_LIMITED = "rate_limited"
+    CONTENT_REJECTED = "content_rejected"
+    SERVER_ERROR = "server_error"
+    UPSTREAM_UNAVAILABLE = "upstream_unavailable"
+
+
+@dataclass(frozen=True)
+class Failure:
+    """一次失败的完整描述。
+
+    :param provider_code: 原始厂商错误码，**原样保留、绝不丢弃**——排查时它往往
+        是唯一线索。领域层不理解它的含义，只负责把它带到日志和响应里。
+    """
+
+    retry: Retry
+    user_facing: UserFacing
+    provider_code: str
+    message: str = ""
+
+
+class TaskState(Enum):
+    """与服务商无关的任务状态。
+
+    刻意比火山的状态集**更小**：火山有 queued / running / cancelled / succeeded /
+    failed / expired 六个，这里把 queued 与 running 合成 PENDING/RUNNING 两态、
+    把 expired 归入 FAILED（它是超时，属于失败的一种，重试策略由 `Failure`
+    单独表达）。映射的取舍放在适配层，领域层只保留上层真正需要区分的状态。
+    """
+
+    PENDING = "pending"
+    RUNNING = "running"
+    SUCCEEDED = "succeeded"
+    FAILED = "failed"
+    CANCELLED = "cancelled"
+
+    @property
+    def is_terminal(self) -> bool:
+        return self in {TaskState.SUCCEEDED, TaskState.FAILED, TaskState.CANCELLED}
+
+
+@dataclass(frozen=True)
+class GenerationRequest:
+    """一次生成请求，服务商无关。
+
+    这是 Java 侧发过来的内容。**刻意不含 `user_id`**：本服务干活只需要
+    `job_id`、生成参数与上传目标，传用户标识属于不必要的扩散（ADR-023 §5
+    数据最小化）。需要给厂商做滥用检测时传的是 `safety_identifier`——一个哈希，
+    不是明文用户 ID。
+
+    :param model: 厂商侧的模型标识，由 Java 侧从目录表的 `provider_model_name`
+        快照取值后传入。**适配层不自行选模型**——选型是目录的职责，且任务行里
+        冻结的快照必须与实际调用的模型一致。
+    :param input_url: 参考图/首帧的公网可达 URL。为空即文生视频（参考图选填，
+        决策 A2）。
+    :param upload_target: 产物的预签名 PUT 地址。当前未启用（转存阻塞于存储
+        决策，ADR-023 §7），保留字段使将来启用无需改契约。
+    """
+
+    job_id: str
+    media_kind: MediaKind
+    model: str
+    prompt: str | None = None
+    input_url: str | None = None
+    ratio: str | None = None
+    resolution: str | None = None
+    duration_sec: int | None = None
+    generate_audio: bool = True
+    watermark: bool = False
+    safety_identifier: str | None = None
+    callback_url: str | None = None
+    upload_target: str | None = None
+
+
+@dataclass(frozen=True)
+class GenerationResult:
+    """一次查询的结果，服务商无关。
+
+    :param asset_url: 产物 URL。**可能是有时效的厂商地址**——火山 Seedance 的
+        有效期是 24 小时，因此调用方必须在有效期内转存，不可长期保存此值。
+        适配层通过 `asset_url_expires_in_sec` 告知时效，避免上层把它当永久地址。
+    :param cover_url: 封面图 URL（视频场景）。火山的尾帧图可免费充当此用途。
+    :param width_px: 像素宽高。**火山不直接返回像素值**，只返回 `ratio` 与
+        `resolution`；换算需要一张随模型family变化的对照表。故此处在 V7 目录
+        种子定下模型族之前保持 None，并把 `ratio`/`resolution` 原样透出，
+        由上层或后续实现换算——编一个可能错的尺寸比留空更糟。
+    :param billing_tokens: 计费对账依据（火山为 `usage.completion_tokens`）。
+    """
+
+    provider_request_id: str
+    state: TaskState
+    progress: int = 0
+    asset_url: str | None = None
+    asset_url_expires_in_sec: int | None = None
+    cover_url: str | None = None
+    ratio: str | None = None
+    resolution: str | None = None
+    width_px: int | None = None
+    height_px: int | None = None
+    duration_ms: int | None = None
+    output_format: str | None = None
+    failure: Failure | None = None
+    billing_tokens: int | None = None
+
+    def __post_init__(self) -> None:
+        # 失败态必须带 Failure，否则上层无法决定重试策略，只能盲目重试或盲目
+        # 放弃——两者都错。宁可在这里炸掉也不要把半个失败态放出去。
+        if self.state is TaskState.FAILED and self.failure is None:
+            raise ValueError("FAILED 状态必须携带 Failure")
+        if self.state is TaskState.SUCCEEDED and not self.asset_url:
+            raise ValueError("SUCCEEDED 状态必须携带 asset_url")

app/providers/base.py

@@ -0,0 +1,49 @@
+"""与服务商无关的生成能力抽象。
+
+上层（HTTP 端点、将来的 Java 侧契约）只对着本接口编程。新增一家供应商 = 实现
+本接口 + 在注册表登记，**不动上层一行**。
+
+为什么是「submit + fetch」两个方法而不是一个 `generate`：
+Java↔Python 边界一律异步（ADR-023 §4）。`submit` 立即返回厂商任务 ID，`fetch`
+查进度。同步型供应商（一次长连接出图）由其适配层在内部消化这个差异，**绝不能
+让它渗透到这个接口上**——否则客户端契约、等待态设计、E2E 都要分两套。
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+from app.domain.generation import GenerationRequest, GenerationResult, MediaKind
+
+
+class GenerationProvider(ABC):
+    """一家供应商的一种生成能力。"""
+
+    @property
+    @abstractmethod
+    def provider_code(self) -> str:
+        """与 `creation.generation_models.provider_code` 同值。
+
+        Java 侧按任务行冻结的 `provider_code_snapshot` 找适配器，**不按当前目录**
+        ——目录可能被改指向或禁用，而在途任务必须仍由当初提交它的适配器轮询。
+        """
+
+    @property
+    @abstractmethod
+    def media_kind(self) -> MediaKind:
+        """本适配器负责的产物类型。"""
+
+    @abstractmethod
+    def submit(self, request: GenerationRequest) -> str:
+        """提交任务，返回厂商任务 ID。
+
+        **不得阻塞等待生成完成。** 调用方会在上游产生任何计费动作之前把返回的 ID
+        落库——那是超时后唯一能判断「是否已提交」的依据。
+
+        :raises app.providers.volcengine.client.ArkError: 或其他供应商的等价异常，
+            均携带已分类的 `Failure`。
+        """
+
+    @abstractmethod
+    def fetch(self, provider_request_id: str) -> GenerationResult:
+        """查询任务当前状态。"""

app/providers/registry.py

@@ -0,0 +1,42 @@
+"""provider 注册表：按 `provider_code` 分发。
+
+Java 侧按任务行冻结的 `provider_code_snapshot` 找适配器，**不按当前目录**——目录
+可能被改指向或禁用，而在途任务必须仍由当初提交它的适配器轮询。用当前目录路由会
+拿着别人的任务 ID 去问错的供应商。
+"""
+
+from __future__ import annotations
+
+from app.providers.base import GenerationProvider
+
+
+class ProviderRegistry:
+    """`provider_code` → 适配器。"""
+
+    def __init__(self, providers: list[GenerationProvider] | None = None):
+        self._by_code: dict[str, GenerationProvider] = {}
+        for provider in providers or []:
+            self.register(provider)
+
+    def register(self, provider: GenerationProvider) -> None:
+        code = provider.provider_code
+        if code in self._by_code:
+            # 重复注册意味着装配出错。静默覆盖会让「到底哪个实例在处理任务」
+            # 取决于注册顺序，是最难查的一类问题。
+            raise ValueError(f"provider_code 重复注册：{code}")
+        self._by_code[code] = provider
+
+    def find(self, provider_code: str) -> GenerationProvider | None:
+        """未知 code 返回 None。
+
+        对一个已持久化的任务而言，未知 code 意味着当初提交它的适配器已被移除。
+        调用方应当作**不可重试**的失败，而不是循环重试到预算耗尽。
+        """
+        return self._by_code.get(provider_code)
+
+    @property
+    def codes(self) -> frozenset[str]:
+        return frozenset(self._by_code)
+
+    def __len__(self) -> int:
+        return len(self._by_code)

app/providers/volcengine/client.py

@@ -0,0 +1,142 @@
+"""火山方舟（Ark）的 HTTP 传输层。
+
+**本模块不认识「视频」这个概念。** 它只管鉴权、超时、请求发送、以及把厂商的错误
+体解包成带分类的异常。以后接火山的图片生成、语音或对话接口时，那些模块复用本
+client，而不是各自再抄一遍鉴权与错误解包——这是把它单独拆出来的全部理由。
+
+判断代码该进这里还是进 `video.py`：问「换成火山的另一个接口，这段还一样吗」。
+鉴权头、错误体形状、超时语义都一样，故在此；`content` 数组怎么拼、`ratio` 怎么
+取值是视频接口特有的，故不在此。
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Self
+
+import httpx
+
+from app.domain.generation import Failure
+
+from .errors import classify
+
+log = logging.getLogger(__name__)
+
+
+class ArkError(RuntimeError):
+    """火山方舟返回的错误，携带领域分类。
+
+    :param failure: 已分类的失败，调用方据此决定重试策略与用户呈现，**不必再去
+        看厂商错误码**——那是适配层的职责边界。
+    """
+
+    def __init__(self, failure: Failure, http_status: int | None = None):
+        super().__init__(failure.message or failure.provider_code or "Ark 调用失败")
+        self.failure = failure
+        self.http_status = http_status
+
+
+class ArkClient:
+    """火山方舟 API 的同步 HTTP 客户端。
+
+    **超时语义**：`timeout_sec` 是单次 HTTP 往返的超时，**不是生成耗时**。方舟的
+    内容生成接口是异步的（提交只入队即返回），所以几秒足够；把它调大成「等出图」
+    会重新引入长连接占用与「不知道有没有扣费」的歧义（ADR-023 §4）。
+    """
+
+    def __init__(
+        self,
+        api_key: str,
+        base_url: str,
+        timeout_sec: float = 10.0,
+        client: httpx.Client | None = None,
+    ):
+        if not api_key:
+            # 缺密钥是服务端配置错误，不是上游故障。在构造点就说清楚，
+            # 免得表现为一串莫名的 401。
+            raise ValueError(
+                "ARK_API_KEY 未配置：请通过环境变量注入，切勿写入代码或样例文件"
+            )
+        self._api_key = api_key
+        self._base_url = base_url.rstrip("/")
+        self._owns_client = client is None
+        self._client = client or httpx.Client(timeout=timeout_sec)
+
+    def post(self, path: str, body: dict[str, Any]) -> dict[str, Any]:
+        return self._send("POST", path, json=body)
+
+    def get(self, path: str) -> dict[str, Any]:
+        return self._send("GET", path)
+
+    # ---------------------------------------------------------------- 内部
+
+    def _send(self, method: str, path: str, **kwargs: Any) -> dict[str, Any]:
+        url = f"{self._base_url}{path}"
+        try:
+            response = self._client.request(
+                method, url, headers=self._headers(), **kwargs
+            )
+        except httpx.HTTPError as exc:
+            # 网络层失败按可重试处理。注意**提交超时属于「不知道有没有扣费」的
+            # 窗口**：火山不支持幂等键，调用方必须依赖已落库的 provider_request_id
+            # 或回调来判断，不可盲目重发。
+            raise ArkError(classify(None, f"请求火山方舟失败：{exc}")) from exc
+        return self._unwrap(response)
+
+    def _headers(self) -> dict[str, str]:
+        return {
+            "Authorization": f"Bearer {self._api_key}",
+            "Content-Type": "application/json",
+        }
+
+    @staticmethod
+    def _unwrap(response: httpx.Response) -> dict[str, Any]:
+        """把响应转成 dict；非 2xx 转成带分类的 `ArkError`。
+
+        错误体形状取自官方错误码表：`{"error": {"code": ..., "message": ...}}`。
+
+        响应体解析失败时**不静默**——那通常意味着打到了错误的地址（例如网关返回
+        HTML 错误页），静默会让排查绕一大圈。
+        """
+        if response.is_success:
+            try:
+                return response.json()
+            except ValueError as exc:
+                raise ArkError(
+                    classify(
+                        None,
+                        f"响应不是合法 JSON（HTTP {response.status_code}）："
+                        f"{response.text[:200]}",
+                    ),
+                    response.status_code,
+                ) from exc
+
+        code: str | None = None
+        message = response.text[:500]
+        try:
+            error = response.json().get("error") or {}
+            code = error.get("code")
+            message = error.get("message") or message
+        except ValueError:
+            pass
+
+        failure = classify(code, message)
+        log.warning(
+            "火山方舟返回错误 http=%s code=%s retry=%s",
+            response.status_code,
+            code,
+            failure.retry.value,
+        )
+        raise ArkError(failure, response.status_code)
+
+    def close(self) -> None:
+        # 只关自己创建的连接池：注入进来的 client 归调用方管理，
+        # 在这里关掉会让共享同一个 client 的其他适配器一起失效。
+        if self._owns_client:
+            self._client.close()
+
+    def __enter__(self) -> Self:
+        return self
+
+    def __exit__(self, *exc_info: object) -> None:
+        self.close()

app/providers/volcengine/errors.py

@@ -0,0 +1,161 @@
+"""火山方舟错误码 → 领域 `Failure` 的映射。
+
+**本模块是唯一理解火山错误码的地方。** 上层只看 `Failure` 的 `retry` 与
+`user_facing`，不接触厂商码——把厂商词汇挡在这一层，是「换供应商不动上层」
+的前提。
+
+三处如果搞错只会在生产暴露的地方：
+
+1. **`QuotaExceeded` 一码两义。** 同一个 code 既表示「免费额度/周期限额耗尽」
+   （要等到明天或去充值），也表示「当前账号排队中任务数超限」（等几秒就好）。
+   官方错误码表里这两条是相邻两行、code 完全相同，只有 message 不同。按 code
+   一刀切，必然有一半场景的文案和出口是错的。
+2. **内容安全类重试毫无意义**（同样的输入必然同样被拒），必须与限流类分开，
+   否则会白烧三次重试预算再失败。
+3. **未知码必须偏向可重试**而不是判死：厂商在持续新增码。
+
+映射的完整取证见 patbond-doc `docs/development/providers/volcengine-seedance.md`。
+"""
+
+from __future__ import annotations
+
+import re
+
+from app.domain.generation import Failure, Retry, UserFacing
+
+#: 内容安全类：**重试无意义**。覆盖输入/输出的文本、图片、视频、音频，以及
+#: 版权、真人、伪造证件等子类，另含风险识别产品（*RiskDetection）的四个码。
+#: 用前缀匹配而非穷举：官方这一族的码在持续增加（Input/Output × 媒体类型 ×
+#: 子原因），穷举必然漏，而这一族的处置方式完全一致。
+_CONTENT_PREFIXES = (
+    "SensitiveContentDetected",
+    "InputTextSensitiveContentDetected",
+    "InputImageSensitiveContentDetected",
+    "InputVideoSensitiveContentDetected",
+    "InputAudioSensitiveContentDetected",
+    "OutputTextSensitiveContentDetected",
+    "OutputImageSensitiveContentDetected",
+    "OutputVideoSensitiveContentDetected",
+    "OutputAudioSensitiveContentDetected",
+    "InputTextRiskDetection",
+    "InputImageRiskDetection",
+    "OutputTextRiskDetection",
+    "OutputImageRiskDetection",
+)
+
+#: 限流类：退避后重试。等的是秒级。
+_RATE_LIMIT_CODES = frozenset(
+    {
+        "RateLimitExceeded.EndpointRPMExceeded",
+        "RateLimitExceeded.EndpointTPMExceeded",
+        "RateLimitExceeded.EndpointFlexTPMExceeded",
+        "ModelAccountRpmRateLimitExceeded",
+        "ModelAccountTpmRateLimitExceeded",
+        "ModelAccountFlexTpmRateLimitExceeded",
+        "ModelAccountIpmRateLimitExceeded",
+        "APIAccountRpmRateLimitExceeded",
+        "AccountRateLimitExceeded",
+        "ServerOverloaded",
+        "RequestBurstTooFast",
+    }
+)
+
+#: 配额/计费类：等的是「明天」或「充值」。重试解决不了。
+#: 注意 403 的欠费与未开通也归此类——HTTP 状态码不同，但用户要做的事一样。
+_QUOTA_CODES = frozenset(
+    {
+        "SetLimitExceeded",
+        "InflightBatchsizeExceeded",
+        "QuotaExceeded.AgentPlanQuotaExceeded",
+        "QuotaExceeded.DoubaoSearchFreeQuotaExceeded",
+        "AccountOverdueError",
+        "OperationDenied.ServiceOverdue",
+        "OperationDenied.FileQuotaExceeded",
+        "ModelNotOpen",
+        "OperationDenied.ServiceNotOpen",
+    }
+)
+
+#: 我方配置或代码错误：重试无意义，需要人来改。
+_SERVER_FAULT_CODES = frozenset(
+    {
+        "AuthenticationError",
+        "InvalidAccountStatus",
+        "AccessDenied",
+        "InvalidEndpointOrModel.NotFound",
+        "InvalidEndpointOrModel.ModelIDAccessDisabled",
+        "InvalidEndpoint.ClosedEndpoint",
+        "OperationDenied.PermissionDenied",
+        "OperationDenied.ArkAccessRoleNotFound",
+        "OperationDenied.TosAccessDenied",
+        "PathNotFound",
+    }
+)
+
+#: 上游内部错误：可重试。
+_UPSTREAM_CODES = frozenset({"InternalServiceError", "UpstreamUnavailable"})
+
+#: `QuotaExceeded` 一码两义的判别式。命中即为「排队任务数超限」，属秒级等待；
+#: 否则按额度耗尽处理。取自官方两行 message 的区别性措辞。
+#: 判别放在 message 上是不得已——code 相同，没有别的字段可依据。
+_QUEUE_BACKLOG_MESSAGE = re.compile(
+    r"exceeded the quota|排队中状态的任务数", re.IGNORECASE
+)
+
+
+def classify(provider_code: str | None, message: str | None = None) -> Failure:
+    """把厂商错误码分类。
+
+    未知码一律按**可退避重试的上游问题**处理而不是直接判死：厂商在持续新增错误
+    码，把未知码当终态失败会让一次可恢复的抖动变成用户看到的失败；而重试预算
+    本身有上限（`max_attempts`），最坏情况只是多试两次。
+    """
+    code = (provider_code or "").strip()
+    text = message or ""
+
+    if not code:
+        return Failure(Retry.BACKOFF, UserFacing.UPSTREAM_UNAVAILABLE, "", text)
+
+    if code.startswith(_CONTENT_PREFIXES):
+        return Failure(Retry.NEVER, UserFacing.CONTENT_REJECTED, code, text)
+
+    if code == "ContentSecurityDetectionError":
+        # 风险识别产品自身请求失败，不是内容真的违规 —— 可重试。
+        return Failure(Retry.BACKOFF, UserFacing.UPSTREAM_UNAVAILABLE, code, text)
+
+    if code == "QuotaExceeded":
+        # 一码两义，见模块 docstring。
+        if _QUEUE_BACKLOG_MESSAGE.search(text):
+            return Failure(Retry.BACKOFF, UserFacing.RATE_LIMITED, code, text)
+        return Failure(Retry.QUOTA, UserFacing.QUOTA_EXHAUSTED, code, text)
+
+    if code in _RATE_LIMIT_CODES:
+        return Failure(Retry.BACKOFF, UserFacing.RATE_LIMITED, code, text)
+
+    if code in _QUOTA_CODES:
+        return Failure(Retry.QUOTA, UserFacing.QUOTA_EXHAUSTED, code, text)
+
+    if code in _SERVER_FAULT_CODES:
+        return Failure(Retry.NEVER, UserFacing.SERVER_ERROR, code, text)
+
+    if code in _UPSTREAM_CODES:
+        return Failure(Retry.BACKOFF, UserFacing.UPSTREAM_UNAVAILABLE, code, text)
+
+    # 参数类：我方拼错了请求体，重试同样的请求体不会变好。
+    # 含 InvalidParameter.TaskTypeConstraint / TaskTypeMismatch —— 那两个是
+    # Seedance 2.5 全模态任务的参数与模型判定的任务类型不兼容，属我方参数问题。
+    if code.startswith(
+        (
+            "InvalidParameter",
+            "MissingParameter",
+            "InvalidArgumentError",
+            "InvalidImageURL",
+            "InvalidPayload",
+            "RequestTooLarge",
+            "RequestBodyTooLarge",
+            "OutofContextError",
+        )
+    ):
+        return Failure(Retry.NEVER, UserFacing.SERVER_ERROR, code, text)
+
+    return Failure(Retry.BACKOFF, UserFacing.UPSTREAM_UNAVAILABLE, code, text)

app/providers/volcengine/models.py

@@ -0,0 +1,187 @@
+"""火山方舟内容生成接口的请求/响应 DTO。
+
+只放**厂商词汇**：字段名、取值、以及厂商文档里那些需要解释的怪异之处。领域概念
+在 `app/domain/generation.py`，两者的翻译在 `video.py`。
+
+这层单独存在的理由：厂商字段名与我们的领域名不一致（`framespersecond` vs `fps`、
+`content.video_url` vs `asset_url`），且厂商有若干**文档自身矛盾或反直觉**之处
+需要就地注明。把这些注释和翻译混进业务代码，业务逻辑就会被厂商细节淹没。
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any
+
+#: 官方 API base URL（北京区）。厂商事实收在厂商包内——放到配置里会让「厂商换
+#: 域名」需要改两处，而其中一处还在与厂商无关的层里。
+DEFAULT_BASE_URL = "https://ark.cn-beijing.volces.com/api/v3"
+
+#: 内容生成任务的路径（提交与查询共用前缀）。
+TASKS_PATH = "/contents/generations/tasks"
+
+#: 产物 URL 的有效期。厂商文档明示 24 小时（Seedance 2.5 另有 100 次下载上限）。
+#: **产物必须在此时限内转存**——把厂商 URL 长期存为 external_url 会变死链。
+ASSET_URL_TTL_SEC = 24 * 3600
+
+
+class SeedanceStatus(Enum):
+    """厂商任务状态。
+
+    **六个取值，而查询接口文档只列了五个。** `expired` 仅见于创建接口的
+    `callback_url` 一节（任务超过 `execution_expires_after` 被自动终止）。只按
+    查询文档实现，线上遇到超时任务会掉进未知分支。
+    """
+
+    QUEUED = "queued"
+    RUNNING = "running"
+    SUCCEEDED = "succeeded"
+    FAILED = "failed"
+    CANCELLED = "cancelled"
+    #: 查询接口文档未列出，来自创建接口的回调状态说明。
+    EXPIRED = "expired"
+
+    @classmethod
+    def parse(cls, value: str | None) -> SeedanceStatus | None:
+        """未知状态返回 None 而非抛异常。
+
+        厂商新增状态时抛异常会让所有在途任务一起失败；返回 None 让上层当成
+        「还没结束」继续轮询，最坏是等到我们自己的超时。
+        """
+        for status in cls:
+            if status.value == value:
+                return status
+        return None
+
+
+@dataclass(frozen=True)
+class SeedanceSubmission:
+    """提交视频生成任务的请求体。
+
+    刻意只暴露 Patbond 会用到的参数。厂商参数远多于此（全模态参考、样片任务、
+    priority、service_tier…），**多暴露一个就多一处要在契约、客户端、文档三处
+    同步维护的地方**，等真用到再加。
+    """
+
+    model: str
+    prompt: str | None = None
+    #: 首帧图片 URL。给了即图生视频，不给即文生视频（参考图选填，决策 A2）。
+    first_frame_url: str | None = None
+    #: `adaptive` 让模型自适应。注意 Seedance 2.5 在首帧/首尾帧、视频编辑、
+    #: 视频延长任务中**只支持 adaptive**，指定具体比例会被拒。
+    ratio: str = "adaptive"
+    resolution: str = "720p"
+    #: 秒。`-1` 交给模型在有效范围内自选（2.5 为 [4,30]，2.0 系列为 [4,15]）。
+    duration: int = 5
+    generate_audio: bool = True
+    watermark: bool = False
+    #: 返回尾帧图。**默认开启**：成本为零，且可直接充当视频封面，省掉抽帧。
+    return_last_frame: bool = True
+    #: 厂商取值范围 [3600, 259200]，默认 48 小时对交互式应用过长。
+    execution_expires_after: int = 3600
+    #: 终端用户哈希标识，供厂商做滥用检测。**必须是哈希，不可传明文用户 ID**。
+    safety_identifier: str | None = None
+    callback_url: str | None = None
+
+    def to_body(self) -> dict[str, Any]:
+        content: list[dict[str, Any]] = []
+        if self.prompt:
+            content.append({"type": "text", "text": self.prompt})
+        if self.first_frame_url:
+            content.append(
+                {
+                    "type": "image_url",
+                    "image_url": {"url": self.first_frame_url},
+                    # role 省略时厂商按 first_frame 处理；显式写出以免依赖默认行为。
+                    "role": "first_frame",
+                }
+            )
+        if not content:
+            # 空内容必然被厂商拒，但那要花一次往返（还可能计入 RPM）。本地先拦。
+            raise ValueError("提交内容为空：prompt 与 first_frame_url 至少要有一个")
+
+        body: dict[str, Any] = {
+            "model": self.model,
+            "content": content,
+            "ratio": self.ratio,
+            "resolution": self.resolution,
+            "duration": self.duration,
+            "generate_audio": self.generate_audio,
+            "watermark": self.watermark,
+            "return_last_frame": self.return_last_frame,
+            "execution_expires_after": self.execution_expires_after,
+        }
+        if self.safety_identifier:
+            body["safety_identifier"] = self.safety_identifier
+        if self.callback_url:
+            body["callback_url"] = self.callback_url
+        return body
+
+
+@dataclass(frozen=True)
+class SeedanceTask:
+    """查询接口返回的任务。
+
+    :param duration_sec: 厂商返回的时长**是向下取整的约数**。官方公式：
+        `duration = 实际总帧数 / 24` 向下取整——133 帧实际 5.54 秒会返回 5。
+    :param frames: 与 `duration_sec` **只会返回其中一个**（提交时指定了 frames
+        才返回 frames）。故 `duration_ms` 优先按帧算。
+    """
+
+    id: str
+    status: SeedanceStatus | None
+    raw_status: str | None
+    video_url: str | None = None
+    last_frame_url: str | None = None
+    duration_sec: int | None = None
+    frames: int | None = None
+    fps: int | None = None
+    ratio: str | None = None
+    resolution: str | None = None
+    output_format: str | None = None
+    error_code: str | None = None
+    error_message: str | None = None
+    completion_tokens: int | None = None
+    created_at: int | None = None
+    updated_at: int | None = None
+
+    @property
+    def duration_ms(self) -> int | None:
+        """尽可能精确的时长毫秒数。
+
+        有 `frames` + `fps` 时按帧算（精确）；否则退回厂商的整数秒（向下取整的
+        约数，最多可能短 1 秒）。两者都缺时返回 None，而不是编一个 0 出来。
+        """
+        if self.frames is not None and self.fps:
+            return round(self.frames / self.fps * 1000)
+        if self.duration_sec is not None:
+            return self.duration_sec * 1000
+        return None
+
+    @classmethod
+    def from_payload(cls, payload: dict[str, Any]) -> SeedanceTask:
+        content = payload.get("content") or {}
+        error = payload.get("error") or {}
+        usage = payload.get("usage") or {}
+        raw_status = payload.get("status")
+        return cls(
+            id=str(payload.get("id") or ""),
+            status=SeedanceStatus.parse(raw_status),
+            raw_status=raw_status,
+            video_url=content.get("video_url"),
+            last_frame_url=content.get("last_frame_url"),
+            duration_sec=payload.get("duration"),
+            frames=payload.get("frames"),
+            # 厂商字段名是 framespersecond（无下划线），不是 fps。
+            fps=payload.get("framespersecond"),
+            ratio=payload.get("ratio"),
+            resolution=payload.get("resolution"),
+            output_format=payload.get("output_format"),
+            error_code=error.get("code"),
+            error_message=error.get("message"),
+            # 视频生成不统计输入 token，故 total == completion，取其一即可。
+            completion_tokens=usage.get("completion_tokens"),
+            created_at=payload.get("created_at"),
+            updated_at=payload.get("updated_at"),
+        )

app/providers/volcengine/video.py

@@ -0,0 +1,193 @@
+"""Seedance 视频生成：领域请求 ↔ 火山请求体的翻译。
+
+本模块只管**视频生成这一个业务**：怎么拼 `content` 数组、状态怎么映射到领域状态、
+失败怎么分类。HTTP 传输与鉴权在 `client.py`，厂商 DTO 在 `models.py`，错误码映射
+在 `errors.py`。
+
+以后接火山的**图片生成**时，应新建一个兄弟模块（如 `image.py`）复用同一个
+`ArkClient` 与 `errors.py`，而不是在本文件里加分支——视频与图片的参数集、状态机、
+产物形态都不同，混在一处会让两边都难改。
+"""
+
+from __future__ import annotations
+
+import logging
+
+from app.domain.generation import (
+    Failure,
+    GenerationRequest,
+    GenerationResult,
+    MediaKind,
+    Retry,
+    TaskState,
+    UserFacing,
+)
+from app.providers.base import GenerationProvider
+
+from .client import ArkClient
+from .models import (
+    ASSET_URL_TTL_SEC,
+    TASKS_PATH,
+    SeedanceStatus,
+    SeedanceSubmission,
+    SeedanceTask,
+)
+
+log = logging.getLogger(__name__)
+
+#: `creation.generation_models.provider_code` 里对应本适配器的值。
+PROVIDER_CODE = "volcengine-seedance"
+
+
+class SeedanceVideoProvider(GenerationProvider):
+    """火山方舟 Seedance 的视频生成适配器。"""
+
+    def __init__(
+        self,
+        client: ArkClient,
+        default_model: str = "",
+        task_expiry_sec: int = 3600,
+    ):
+        self._client = client
+        self._default_model = default_model
+        self._task_expiry_sec = task_expiry_sec
+
+    @property
+    def provider_code(self) -> str:
+        return PROVIDER_CODE
+
+    @property
+    def media_kind(self) -> MediaKind:
+        return MediaKind.VIDEO
+
+    # ------------------------------------------------------------- submit
+
+    def submit(self, request: GenerationRequest) -> str:
+        if request.media_kind is not MediaKind.VIDEO:
+            # 路由错了。与其把一个图片请求塞给视频接口让厂商拒，不如就地说清楚。
+            raise ValueError(
+                f"{PROVIDER_CODE} 只处理视频生成，收到 {request.media_kind.value}"
+            )
+
+        submission = self._to_submission(request)
+        payload = self._client.post(TASKS_PATH, submission.to_body())
+        task_id = payload.get("id")
+        if not task_id:
+            # 2xx 但没有 ID：我们再也查不到这个任务，而它可能已经在计费。
+            # 绝不能当成功放过。
+            raise ValueError(f"火山方舟提交成功但响应缺少任务 ID：{payload!r}")
+        return str(task_id)
+
+    def _to_submission(self, request: GenerationRequest) -> SeedanceSubmission:
+        """领域请求 → 厂商请求体。
+
+        模型取值优先用请求里的（来自任务行冻结的快照），仅在缺失时回落到配置的
+        默认值。**适配层不自行选模型**：选型是目录的职责，且任务行里的快照必须
+        与实际调用的模型一致，否则事后无法对账。
+        """
+        model = request.model or self._default_model
+        if not model:
+            raise ValueError("未指定模型：请求未携带 model 且 ARK_VIDEO_MODEL 未配置")
+        return SeedanceSubmission(
+            model=model,
+            prompt=request.prompt,
+            first_frame_url=request.input_url,
+            ratio=request.ratio or SeedanceSubmission.ratio,
+            resolution=request.resolution or SeedanceSubmission.resolution,
+            duration=(
+                request.duration_sec
+                if request.duration_sec is not None
+                else SeedanceSubmission.duration
+            ),
+            generate_audio=request.generate_audio,
+            watermark=request.watermark,
+            execution_expires_after=self._task_expiry_sec,
+            safety_identifier=request.safety_identifier,
+            callback_url=request.callback_url,
+        )
+
+    # -------------------------------------------------------------- fetch
+
+    def fetch(self, provider_request_id: str) -> GenerationResult:
+        payload = self._client.get(f"{TASKS_PATH}/{provider_request_id}")
+        task = SeedanceTask.from_payload(payload)
+        return self._to_result(task, provider_request_id)
+
+    def _to_result(self, task: SeedanceTask, request_id: str) -> GenerationResult:
+        """厂商任务 → 领域结果。
+
+        状态映射见下。注意领域状态集比厂商的**更小**：queued/running 都还没结束，
+        expired 是失败的一种（重试策略由 `Failure` 表达，不需要一个独立状态）。
+        """
+        state = self._map_state(task)
+        failure = self._map_failure(task)
+
+        if state is TaskState.SUCCEEDED and not task.video_url:
+            # 厂商说成功却没给 URL：这不是成功。当成可重试的上游异常，
+            # 而不是把一个没有产物的「成功」放给上层去空指针。
+            log.error("火山方舟任务 %s 状态为 succeeded 但缺少 video_url", task.id)
+            return GenerationResult(
+                provider_request_id=request_id,
+                state=TaskState.FAILED,
+                failure=Failure(
+                    Retry.BACKOFF,
+                    UserFacing.UPSTREAM_UNAVAILABLE,
+                    "MissingVideoUrl",
+                    "厂商返回 succeeded 但 content.video_url 为空",
+                ),
+            )
+
+        return GenerationResult(
+            provider_request_id=request_id,
+            state=state,
+            # 火山不返回百分比进度。这不是缺陷而是事实：客户端契约本就不承诺
+            # 平滑进度（ADR-023 §10），成功前一律 0、成功即 100。
+            progress=100 if state is TaskState.SUCCEEDED else 0,
+            asset_url=task.video_url,
+            asset_url_expires_in_sec=(ASSET_URL_TTL_SEC if task.video_url else None),
+            cover_url=task.last_frame_url,
+            ratio=task.ratio,
+            resolution=task.resolution,
+            # 像素尺寸留空：火山只返回 ratio 与 resolution，换算需要一张随模型族
+            # 变化的对照表。在 V7 目录种子定下模型族之前不编造尺寸。
+            duration_ms=task.duration_ms,
+            output_format=task.output_format,
+            failure=failure,
+            billing_tokens=task.completion_tokens,
+        )
+
+    @staticmethod
+    def _map_state(task: SeedanceTask) -> TaskState:
+        match task.status:
+            case SeedanceStatus.QUEUED:
+                return TaskState.PENDING
+            case SeedanceStatus.RUNNING:
+                return TaskState.RUNNING
+            case SeedanceStatus.SUCCEEDED:
+                return TaskState.SUCCEEDED
+            case SeedanceStatus.CANCELLED:
+                return TaskState.CANCELLED
+            case SeedanceStatus.FAILED | SeedanceStatus.EXPIRED:
+                return TaskState.FAILED
+            case _:
+                # 未知状态当作「还在跑」：厂商新增状态时把在途任务判死更糟，
+                # 最坏是等到我们自己的超时。
+                log.warning("火山方舟返回未知任务状态：%r", task.raw_status)
+                return TaskState.RUNNING
+
+    @staticmethod
+    def _map_failure(task: SeedanceTask) -> Failure | None:
+        from .errors import classify
+
+        if task.status is SeedanceStatus.EXPIRED:
+            # 超时不是内容或参数问题，可重试。单独处理是因为厂商在 expired 时
+            # 不一定给 error.code，走通用分类会被当成「无码」。
+            return Failure(
+                Retry.BACKOFF,
+                UserFacing.UPSTREAM_UNAVAILABLE,
+                "TaskExpired",
+                "任务超过 execution_expires_after 被厂商终止",
+            )
+        if task.status is SeedanceStatus.FAILED:
+            return classify(task.error_code, task.error_message)
+        return None

requirements-dev.txt

@@ -0,0 +1,4 @@
+-r requirements.txt
+# 与 CI 门禁同一套工具（.gitea/workflows/ci.yml）
+ruff>=0.6
+pytest>=8.0

requirements.txt

@@ -0,0 +1,3 @@
+# 运行时依赖。版本下限按实测可用版本给，不锁死上限——本服务只用 httpx 的
+# 基础同步 API，其小版本变更风险低。
+httpx>=0.27

scripts/check-secrets.sh

@@ -0,0 +1,119 @@
+#!/bin/sh
+# check-secrets.sh —— 凭证防泄漏检查（ADR-021，规范见 patbond-doc docs/development/git-workflow.md）
+#
+# 规则单一来源：本地 pre-commit 与 CI 兜底跑的是同一个脚本、同一张规则表。
+# 四仓（patbond-api / patbond-flutter / patbond-doc / patbond-ai）各存一份同构副本，
+# 改规则时四仓同步提交。patbond-ai 是唯一持有 AI 服务商凭证的仓，尤其不可漏。
+#
+# 用法：
+#   sh scripts/check-secrets.sh --staged    # pre-commit：扫暂存区内容（经 scripts/hooks/pre-commit 调用）
+#   sh scripts/check-secrets.sh --all       # CI 兜底 / 手动自查：扫全部已跟踪文件（缺省模式）
+#   sh scripts/check-secrets.sh <文件...>   # 扫指定文件
+#
+# 拦下真实凭证时的第一动作：去云控制台轮换/禁用该密钥，然后才是清理提交。
+set -u
+
+mode="${1:---all}"
+
+# 允许清单：行内出现任一形态即放行（${} 注入、占位值、明显示例值）
+# 注：x{3,}（xxx / xxxx…）是本项目文档描述凭证形态时的既有占位约定，
+# 放行它使「讲解规则的文档」不会被规则自身拦下（如 iteration-4/08 §漏检清单）。
+ALLOW='\$\{[^}]*\}|\{\{[^}]*\}\}|changeme|change[-_]me|your[-_][a-zA-Z0-9_-]+|<[a-zA-Z0-9 ,_.-]+>|placeholder|example|sample|dummy|fake|redacted|\*\*\*|x{3,}'
+
+# 依赖注入的传参不是字面量密钥：`api_key=settings.api_key`、
+# `.param("secretKey", props.getSecretKey())` 这类行里没有任何秘密，而做 DI 的
+# 代码到处都是这种写法——不放行就得在每个调用点绕过一次。
+# 收得很紧，避免把真凭证放进来：
+#   * 值必须是**未加引号**的小写起首标识符引用（真密钥要么带引号，要么不是标识符形状）
+#   * 只允许**恰好一个点**且随即行尾（JWT 是 a.b.c 三段，仍会被拦）
+#   * 连字符不在字符集内（sk-ant-… 这类前缀仍会被拦）
+ALLOW_REFERENCE='[:=][[:space:]]*[a-z][a-zA-Z0-9_]*\.[a-z][a-zA-Z0-9_]*[[:space:]]*[,);]?$'
+
+# 内容扫描跳过：本脚本与 hook 自身（含规则文本，非凭证）
+SKIP_PATHS='(^|/)scripts/(check-secrets\.sh|hooks/pre-commit)$'
+
+# 文件名黑名单：凭证载体文件本体禁止入库（.sample/.example 除外）
+DENY_NAME='(^|/)\.env(\.[^/]+)?$|(^|/)credentials[^/]*$|[Aa]ccess[Kk]eys?[^/]*\.csv$|(^|/)rootkey\.csv$'
+DENY_NAME_OK='\.(sample|example)$'
+
+# 规则表：ID<TAB>大小写旗标(i=忽略大小写，-=敏感)<TAB>文件范围ERE(-=全部文件)<TAB>行模式ERE
+RULES=$(cat <<'EOF'
+AK-AWS	-	-	AKIA[0-9A-Z]{16}
+AK-QCLOUD	-	-	AKID[0-9A-Za-z]{16,}
+AK-ALIYUN	-	-	LTAI[0-9A-Za-z]{12,}
+AK-ANTHROPIC	-	-	sk-ant-[A-Za-z0-9_-]{16,}
+AK-OPENAI	-	-	sk-(proj-|svcacct-)?[A-Za-z0-9]{20,}
+MINIO-DEFAULT	i	-	minio[-_.]?admin
+PRIVATE-KEY	-	-	^[[:space:]]*-----BEGIN [A-Z ]*PRIVATE KEY-----[[:space:]]*$
+KEY-ASSIGN	i	-	(access[-_]?key(_?id)?|secret[-_]?(access[-_]?)?key|api[-_]?(key|secret|token)|auth[-_]?token)["']?[[:space:]]*[:=][[:space:]]*["']?[A-Za-z0-9+/=_-]{8,}
+JWT-SECRET	i	-	(jwt[-_.]?secret|signing[-_]?key|token[-_]?secret|hmac[-_]?(key|secret))["']?[[:space:]]*[:=][[:space:]]*["']?[A-Za-z0-9+/=_-]{8,}
+DB-PASSWORD	i	\.(ya?ml|properties|toml|conf|ini)(\.sample|\.example)?$	(password|passwd|pwd)["']?[[:space:]]*[:=][[:space:]]*["']?[^[:space:]"'$]{6,}
+EOF
+)
+
+case "$mode" in
+--staged)
+    files=$(git diff --cached --name-only --diff-filter=ACM)
+    src=index
+    ;;
+--all)
+    files=$(git ls-files)
+    src=worktree
+    ;;
+-*)
+    echo "用法: $0 [--staged|--all|<文件...>]" >&2
+    exit 2
+    ;;
+*)
+    files=$(printf '%s\n' "$@")
+    src=worktree
+    ;;
+esac
+
+[ -n "$files" ] || exit 0
+
+tmp=$(mktemp) || exit 2
+viol=$(mktemp) || exit 2
+trap 'rm -f "$tmp" "$viol"' EXIT
+
+# 第一道：文件名黑名单
+printf '%s\n' "$files" | grep -E "$DENY_NAME" | grep -vE "$DENY_NAME_OK" |
+    sed 's/^/[NAME-DENY] /' >>"$viol" || true
+
+# 第二道：逐文件逐规则内容扫描（二进制文件经 grep -I 自然跳过）
+IFS='
+'
+for f in $files; do
+    printf '%s' "$f" | grep -qE "$SKIP_PATHS" && continue
+    if [ "$src" = index ]; then
+        git show ":$f" >"$tmp" 2>/dev/null || continue
+    else
+        [ -f "$f" ] || continue
+        cat -- "$f" >"$tmp"
+    fi
+    printf '%s\n' "$RULES" | while IFS="$(printf '\t')" read -r id flag scope pat; do
+        [ -n "$id" ] || continue
+        if [ "$scope" != "-" ]; then
+            printf '%s' "$f" | grep -qE "$scope" || continue
+        fi
+        ci=""
+        [ "$flag" = "i" ] && ci="-i"
+        grep -InE $ci -e "$pat" "$tmp" 2>/dev/null | grep -viE "$ALLOW" |
+            grep -vE "$ALLOW_REFERENCE" |
+            sed "s|^|[$id] $f:|" >>"$viol" || true
+    done
+done
+
+if [ -s "$viol" ]; then
+    echo "凭证防泄漏检查未通过（ADR-021）——以下内容疑似真实凭证：" >&2
+    cat "$viol" >&2
+    cat >&2 <<'MSG'
+处置：
+  1. 若是真实凭证：先去云控制台轮换/禁用该密钥，再从提交中移除；
+  2. 若是误报：改用 ${} 注入或占位值（changeme / your-xxx / <占位>），
+     或与团队确认后调整四仓同构的 scripts/check-secrets.sh 规则表。
+敏感信息只允许存在于被 gitignore 的文件或 .sample 占位中（git-workflow.md）。
+MSG
+    exit 1
+fi
+exit 0

scripts/hooks/pre-commit

@@ -0,0 +1,6 @@
+#!/bin/sh
+# pre-commit —— 凭证防泄漏（ADR-021）。启用（每人每仓一次）：
+#   git config core.hooksPath scripts/hooks
+# 注意：core.hooksPath 会整体接管 hooks 目录；本仓无其他自定义 hook。
+repo_root=$(git rev-parse --show-toplevel) || exit 1
+exec sh "$repo_root/scripts/check-secrets.sh" --staged

tests/providers/volcengine/test_client.py

@@ -0,0 +1,153 @@
+"""`ArkClient` 的测试——只测传输层：鉴权头、错误解包、连接生命周期。
+
+**刻意不测任何视频业务**：本 client 不认识「视频」，测试也不该。视频相关的断言
+在 `test_video.py`。
+"""
+
+from __future__ import annotations
+
+import httpx
+import pytest
+
+from app.domain.generation import Retry, UserFacing
+from app.providers.volcengine.client import ArkClient, ArkError
+
+#: 显式的假值。防泄漏扫描会拦下 `api_key = "..."` 形态（规则正确——这正是真实
+#: 密钥最常见的泄漏形态），故夹具用 ALLOW 清单认可的占位词。
+FAKE_API_KEY = "dummy-api-key-for-tests"
+
+BASE_URL = "https://ark.example.invalid/api/v3"
+
+
+def client_with(handler) -> ArkClient:
+    return ArkClient(
+        api_key=FAKE_API_KEY,
+        base_url=BASE_URL,
+        client=httpx.Client(transport=httpx.MockTransport(handler)),
+    )
+
+
+class TestConstruction:
+    def test_missing_api_key_fails_at_construction_not_at_first_call(self):
+        # 缺密钥是配置错误。让它表现为一串莫名的 401 会浪费排查时间。
+        with pytest.raises(ValueError, match="ARK_API_KEY"):
+            ArkClient(api_key="", base_url=BASE_URL)
+
+    def test_trailing_slash_in_base_url_does_not_double_up(self):
+        def handler(request: httpx.Request) -> httpx.Response:
+            assert "//contents" not in str(request.url)
+            return httpx.Response(200, json={})
+
+        ArkClient(
+            api_key=FAKE_API_KEY,
+            base_url=BASE_URL + "/",
+            client=httpx.Client(transport=httpx.MockTransport(handler)),
+        ).get("/contents/x")
+
+
+class TestAuthAndTransport:
+    def test_bearer_token_and_content_type_are_sent(self):
+        def handler(request: httpx.Request) -> httpx.Response:
+            assert request.headers["Authorization"] == f"Bearer {FAKE_API_KEY}"
+            assert request.headers["Content-Type"] == "application/json"
+            return httpx.Response(200, json={"ok": True})
+
+        assert client_with(handler).post("/x", {}) == {"ok": True}
+
+    def test_get_and_post_hit_the_expected_method(self):
+        seen: list[str] = []
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            seen.append(request.method)
+            return httpx.Response(200, json={})
+
+        client = client_with(handler)
+        client.post("/x", {})
+        client.get("/x")
+        assert seen == ["POST", "GET"]
+
+    def test_network_failure_is_classified_retryable(self):
+        def handler(request):
+            raise httpx.ConnectError("connection refused")
+
+        with pytest.raises(ArkError) as exc:
+            client_with(handler).get("/x")
+        assert exc.value.failure.retry is Retry.BACKOFF
+        assert exc.value.http_status is None
+
+
+class TestErrorUnwrapping:
+    def test_vendor_error_body_is_classified(self):
+        def handler(request):
+            return httpx.Response(
+                429,
+                json={
+                    "error": {
+                        "code": "RateLimitExceeded.EndpointRPMExceeded",
+                        "message": "RPM limit exceeded",
+                    }
+                },
+            )
+
+        with pytest.raises(ArkError) as exc:
+            client_with(handler).get("/x")
+        assert exc.value.failure.retry is Retry.BACKOFF
+        assert exc.value.failure.user_facing is UserFacing.RATE_LIMITED
+        assert (
+            exc.value.failure.provider_code == "RateLimitExceeded.EndpointRPMExceeded"
+        )
+        assert exc.value.http_status == 429
+
+    def test_vendor_message_is_preserved_for_troubleshooting(self):
+        def handler(request):
+            return httpx.Response(
+                401,
+                json={
+                    "error": {"code": "AuthenticationError", "message": "key invalid"}
+                },
+            )
+
+        with pytest.raises(ArkError) as exc:
+            client_with(handler).get("/x")
+        assert "key invalid" in exc.value.failure.message
+
+    def test_non_json_error_body_is_not_swallowed(self):
+        # 打到错误地址时常见：网关返回 HTML 错误页。静默会让排查绕远路。
+        def handler(request):
+            return httpx.Response(502, text="<html>bad gateway</html>")
+
+        with pytest.raises(ArkError) as exc:
+            client_with(handler).get("/x")
+        assert "bad gateway" in exc.value.failure.message
+        assert exc.value.failure.retry is Retry.BACKOFF
+
+    def test_non_json_success_body_is_an_error_too(self):
+        def handler(request):
+            return httpx.Response(200, text="not json")
+
+        with pytest.raises(ArkError, match="不是合法 JSON"):
+            client_with(handler).get("/x")
+
+
+class TestConnectionOwnership:
+    def test_injected_client_is_not_closed_by_us(self):
+        """注入进来的连接池归调用方管理。
+
+        在这里关掉会让共享同一个 httpx.Client 的其他适配器一起失效——接了第二个
+        火山接口后这会变成一个极难查的偶发失败。
+        """
+        injected = httpx.Client(
+            transport=httpx.MockTransport(lambda r: httpx.Response(200, json={}))
+        )
+        ArkClient(api_key=FAKE_API_KEY, base_url=BASE_URL, client=injected).close()
+        assert not injected.is_closed
+
+    def test_own_client_is_closed(self):
+        client = ArkClient(api_key=FAKE_API_KEY, base_url=BASE_URL)
+        client.close()
+        assert client._client.is_closed
+
+    def test_usable_as_context_manager(self):
+        with ArkClient(api_key=FAKE_API_KEY, base_url=BASE_URL) as client:
+            inner = client._client
+        assert inner.is_closed

tests/providers/volcengine/test_errors.py

@@ -0,0 +1,176 @@
+"""火山方舟错误分类的测试。
+
+每个用例对应官方错误码表里的一行或一类。重点在三处**如果搞错只会在生产暴露**
+的地方：`QuotaExceeded` 的一码两义、内容安全类必须不可重试、未知码必须偏向
+可重试而不是直接判死。
+"""
+
+from app.domain.generation import Retry, UserFacing
+from app.providers.volcengine.errors import classify
+
+
+class TestQuotaExceededIsTwoDifferentThings:
+    """同一个 code，两种完全不同的处置。只看 code 必然错一半。"""
+
+    def test_queue_backlog_is_a_seconds_long_wait(self):
+        # 官方原文：「当前账号处于排队中状态的任务数已超过限制，请稍后重试。」
+        result = classify(
+            "QuotaExceeded", "The request has exceeded the quota. Request ID: x."
+        )
+        assert result.retry is Retry.BACKOFF
+        assert result.user_facing is UserFacing.RATE_LIMITED
+
+    def test_queue_backlog_recognised_from_chinese_message(self):
+        result = classify("QuotaExceeded", "当前账号处于排队中状态的任务数已超过限制")
+        assert result.retry is Retry.BACKOFF
+        assert result.user_facing is UserFacing.RATE_LIMITED
+
+    def test_free_trial_exhausted_needs_billing_action(self):
+        # 官方原文：「免费试用额度已消耗完毕，请前往控制台开通对应模型服务。」
+        result = classify(
+            "QuotaExceeded",
+            "Your account [x] has exhausted its free trial quota for the [y] model.",
+        )
+        assert result.retry is Retry.QUOTA
+        assert result.user_facing is UserFacing.QUOTA_EXHAUSTED
+
+    def test_periodic_limit_exhausted_needs_to_wait_for_reset(self):
+        result = classify(
+            "QuotaExceeded",
+            "You have exceeded the 5-hour/weekly/monthly usage quota. It will reset at T.",
+        )
+        assert result.retry is Retry.QUOTA
+        assert result.user_facing is UserFacing.QUOTA_EXHAUSTED
+
+
+class TestContentRejectionMustNotBeRetried:
+    """同样的输入必然同样被拒。重试只会白烧额度再失败。"""
+
+    def test_input_image_sensitive(self):
+        result = classify("InputImageSensitiveContentDetected")
+        assert result.retry is Retry.NEVER
+        assert result.user_facing is UserFacing.CONTENT_REJECTED
+
+    def test_output_video_sensitive(self):
+        result = classify("OutputVideoSensitiveContentDetected")
+        assert result.retry is Retry.NEVER
+        assert result.user_facing is UserFacing.CONTENT_REJECTED
+
+    def test_policy_violation_subcode(self):
+        result = classify("InputVideoSensitiveContentDetected.PolicyViolation")
+        assert result.retry is Retry.NEVER
+
+    def test_real_person_subcode(self):
+        result = classify("InputImageSensitiveContentDetected.PrivacyInformation")
+        assert result.retry is Retry.NEVER
+
+    def test_risk_detection_product_codes(self):
+        assert classify("InputTextRiskDetection").retry is Retry.NEVER
+        assert classify("OutputImageRiskDetection").retry is Retry.NEVER
+
+    def test_but_the_detector_itself_failing_is_retryable(self):
+        # 风险识别产品请求失败 ≠ 内容真的违规。
+        result = classify("ContentSecurityDetectionError", "Internal error.")
+        assert result.retry is Retry.BACKOFF
+        assert result.user_facing is UserFacing.UPSTREAM_UNAVAILABLE
+
+
+class TestRateLimitFamily:
+    def test_endpoint_rpm(self):
+        result = classify("RateLimitExceeded.EndpointRPMExceeded")
+        assert result.retry is Retry.BACKOFF
+        assert result.user_facing is UserFacing.RATE_LIMITED
+
+    def test_model_account_tpm(self):
+        assert classify("ModelAccountTpmRateLimitExceeded").retry is Retry.BACKOFF
+
+    def test_server_overloaded(self):
+        assert classify("ServerOverloaded").retry is Retry.BACKOFF
+
+    def test_request_burst_too_fast(self):
+        # doubao-seed-2.0 之后版本的突增流量保护码。
+        assert classify("RequestBurstTooFast").retry is Retry.BACKOFF
+
+    def test_images_per_minute(self):
+        assert classify("ModelAccountIpmRateLimitExceeded").retry is Retry.BACKOFF
+
+
+class TestQuotaAndBillingFamily:
+    def test_inference_limit_set_by_account(self):
+        result = classify("SetLimitExceeded")
+        assert result.retry is Retry.QUOTA
+        assert result.user_facing is UserFacing.QUOTA_EXHAUSTED
+
+    def test_concurrency_cap_needs_topup(self):
+        assert classify("InflightBatchsizeExceeded").retry is Retry.QUOTA
+
+    def test_account_overdue_is_403_but_same_user_action(self):
+        # HTTP 状态码不同（403 而非 429），但用户要做的事一样：去充值。
+        result = classify("AccountOverdueError")
+        assert result.retry is Retry.QUOTA
+        assert result.user_facing is UserFacing.QUOTA_EXHAUSTED
+
+    def test_model_not_activated(self):
+        assert classify("ModelNotOpen").retry is Retry.QUOTA
+
+
+class TestOurOwnFaults:
+    """重试无意义，需要人改代码或改配置。"""
+
+    def test_bad_api_key(self):
+        result = classify("AuthenticationError")
+        assert result.retry is Retry.NEVER
+        assert result.user_facing is UserFacing.SERVER_ERROR
+
+    def test_unknown_model_id(self):
+        assert classify("InvalidEndpointOrModel.NotFound").retry is Retry.NEVER
+
+    def test_invalid_parameter(self):
+        assert classify("InvalidParameter").retry is Retry.NEVER
+
+    def test_task_type_constraint_is_our_parameter_problem(self):
+        # Seedance 2.5 全模态任务的参数与模型判定的任务类型不兼容。
+        assert classify("InvalidParameter.TaskTypeConstraint").retry is Retry.NEVER
+
+    def test_task_type_mismatch(self):
+        assert classify("InvalidParameter.TaskTypeMismatch").retry is Retry.NEVER
+
+    def test_missing_parameter(self):
+        assert classify("MissingParameter").retry is Retry.NEVER
+
+    def test_payload_too_large(self):
+        assert classify("RequestTooLarge").retry is Retry.NEVER
+
+
+class TestUpstreamErrors:
+    def test_internal_service_error_is_retryable(self):
+        result = classify("InternalServiceError")
+        assert result.retry is Retry.BACKOFF
+        assert result.user_facing is UserFacing.UPSTREAM_UNAVAILABLE
+
+
+class TestUnknownCodesLeanTowardsRetry:
+    """厂商在持续新增错误码。把未知码判死会让一次抖动变成用户看到的失败；
+    重试预算本身有上限，最坏只是多试两次。"""
+
+    def test_unknown_code(self):
+        result = classify("SomeBrandNewCodeVolcengineAddedLastWeek")
+        assert result.retry is Retry.BACKOFF
+        assert result.user_facing is UserFacing.UPSTREAM_UNAVAILABLE
+
+    def test_missing_code(self):
+        assert classify(None).retry is Retry.BACKOFF
+        assert classify("").retry is Retry.BACKOFF
+
+
+class TestProviderCodeIsAlwaysPreserved:
+    """原始码绝不丢弃——排查时它是唯一线索。"""
+
+    def test_known_code(self):
+        assert classify("ServerOverloaded").provider_code == "ServerOverloaded"
+
+    def test_unknown_code(self):
+        assert classify("WeirdCode").provider_code == "WeirdCode"
+
+    def test_whitespace_is_trimmed(self):
+        assert classify("  ServerOverloaded  ").provider_code == "ServerOverloaded"

tests/providers/volcengine/test_video.py

@@ -0,0 +1,398 @@
+"""Seedance 视频适配器的测试：领域请求 ↔ 厂商请求体的翻译，以及状态映射。
+
+夹具里的响应体逐字取自官方文档的「返回格式」示例，避免测试通过而真实字段名对不上。
+"""
+
+from __future__ import annotations
+
+import httpx
+import pytest
+
+from app.domain.generation import (
+    GenerationRequest,
+    MediaKind,
+    Retry,
+    TaskState,
+    UserFacing,
+)
+from app.providers.volcengine.client import ArkClient
+from app.providers.volcengine.models import SeedanceSubmission
+from app.providers.volcengine.video import PROVIDER_CODE, SeedanceVideoProvider
+
+FAKE_API_KEY = "dummy-api-key-for-tests"
+
+#: 官方查询文档「返回格式」一节的原样示例。
+OFFICIAL_SUCCESS_PAYLOAD = {
+    "id": "cgt-2026****-****",
+    "model": "doubao-seedance-2-0-260128",
+    "status": "succeeded",
+    "content": {
+        "video_url": "https://ark-content-generation-cn-beijing.tos-cn-beijing.volces.com/xxx"
+    },
+    "usage": {"completion_tokens": 108900, "total_tokens": 108900},
+    "created_at": 1779348818,
+    "updated_at": 1779348874,
+    "seed": 78674,
+    "resolution": "720p",
+    "ratio": "16:9",
+    "duration": 5,
+    "framespersecond": 24,
+    "service_tier": "default",
+    "execution_expires_after": 172800,
+    "generate_audio": True,
+    "draft": False,
+    "priority": 0,
+}
+
+
+def provider_with(handler, **kwargs) -> SeedanceVideoProvider:
+    client = ArkClient(
+        api_key=FAKE_API_KEY,
+        base_url="https://ark.example.invalid/api/v3",
+        client=httpx.Client(transport=httpx.MockTransport(handler)),
+    )
+    return SeedanceVideoProvider(client, **kwargs)
+
+
+def video_request(**kwargs) -> GenerationRequest:
+    defaults = {
+        "job_id": "job-1",
+        "media_kind": MediaKind.VIDEO,
+        "model": "doubao-seedance-x",
+        "prompt": "一只猫",
+    }
+    return GenerationRequest(**{**defaults, **kwargs})
+
+
+class TestIdentity:
+    def test_provider_code_matches_the_catalog_column(self):
+        assert (
+            provider_with(lambda r: httpx.Response(200, json={})).provider_code
+            == PROVIDER_CODE
+        )
+
+    def test_declares_itself_as_a_video_provider(self):
+        assert (
+            provider_with(lambda r: httpx.Response(200, json={})).media_kind
+            is MediaKind.VIDEO
+        )
+
+
+class TestSubmitTranslation:
+    def test_text_to_video_sends_only_text(self):
+        captured: dict = {}
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            import json
+
+            captured.update(json.loads(request.content))
+            return httpx.Response(200, json={"id": "cgt-1"})
+
+        provider_with(handler).submit(video_request())
+        assert captured["content"] == [{"type": "text", "text": "一只猫"}]
+
+    def test_input_url_becomes_first_frame_with_explicit_role(self):
+        captured: dict = {}
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            import json
+
+            captured.update(json.loads(request.content))
+            return httpx.Response(200, json={"id": "cgt-1"})
+
+        provider_with(handler).submit(video_request(input_url="https://x/a.png"))
+        image = captured["content"][1]
+        assert image["type"] == "image_url"
+        assert image["image_url"]["url"] == "https://x/a.png"
+        # role 省略时厂商按 first_frame 处理，但显式写出以免依赖默认行为。
+        assert image["role"] == "first_frame"
+
+    def test_model_from_request_wins_over_the_configured_default(self):
+        """请求里的模型来自任务行冻结的快照，必须优先。
+
+        否则「任务记录里写的模型」与「实际调用的模型」会不一致，事后无法对账。
+        """
+        captured: dict = {}
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            import json
+
+            captured.update(json.loads(request.content))
+            return httpx.Response(200, json={"id": "cgt-1"})
+
+        provider_with(handler, default_model="fallback-model").submit(
+            video_request(model="snapshot-model")
+        )
+        assert captured["model"] == "snapshot-model"
+
+    def test_configured_default_is_only_a_fallback(self):
+        captured: dict = {}
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            import json
+
+            captured.update(json.loads(request.content))
+            return httpx.Response(200, json={"id": "cgt-1"})
+
+        provider_with(handler, default_model="fallback-model").submit(
+            video_request(model="")
+        )
+        assert captured["model"] == "fallback-model"
+
+    def test_no_model_anywhere_fails_before_calling_the_vendor(self):
+        with pytest.raises(ValueError, match="未指定模型"):
+            provider_with(lambda r: httpx.Response(200, json={})).submit(
+                video_request(model="")
+            )
+
+    def test_task_expiry_comes_from_config_not_the_vendor_default(self):
+        # 厂商默认 48 小时，对交互式应用过长——被放弃的任务会在厂商侧挂两天。
+        captured: dict = {}
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            import json
+
+            captured.update(json.loads(request.content))
+            return httpx.Response(200, json={"id": "cgt-1"})
+
+        provider_with(handler, task_expiry_sec=7200).submit(video_request())
+        assert captured["execution_expires_after"] == 7200
+
+    def test_rejects_a_misrouted_image_request(self):
+        with pytest.raises(ValueError, match="只处理视频生成"):
+            provider_with(lambda r: httpx.Response(200, json={})).submit(
+                video_request(media_kind=MediaKind.IMAGE)
+            )
+
+    def test_returns_vendor_task_id(self):
+        def handler(request):
+            return httpx.Response(200, json={"id": "cgt-123"})
+
+        assert provider_with(handler).submit(video_request()) == "cgt-123"
+
+    def test_success_without_task_id_is_an_error(self):
+        # 2xx 但没有 ID：我们再也查不到这个任务，而它可能已经在计费。
+        def handler(request):
+            return httpx.Response(200, json={"unexpected": True})
+
+        with pytest.raises(ValueError, match="缺少任务 ID"):
+            provider_with(handler).submit(video_request())
+
+
+class TestSubmissionBody:
+    """`SeedanceSubmission` 自身的行为，不经过 HTTP。"""
+
+    def test_empty_content_is_rejected_locally(self):
+        # 空内容必然被厂商拒，但那要花一次往返（还可能计入 RPM）。
+        with pytest.raises(ValueError, match="至少要有一个"):
+            SeedanceSubmission(model="m").to_body()
+
+    def test_last_frame_requested_by_default_for_a_free_cover_image(self):
+        assert (
+            SeedanceSubmission(model="m", prompt="x").to_body()["return_last_frame"]
+            is True
+        )
+
+    def test_optional_fields_are_omitted_when_unset(self):
+        body = SeedanceSubmission(model="m", prompt="x").to_body()
+        assert "safety_identifier" not in body
+        assert "callback_url" not in body
+
+
+class TestFetchStateMapping:
+    def test_succeeded(self):
+        result = provider_with(
+            lambda r: httpx.Response(200, json=OFFICIAL_SUCCESS_PAYLOAD)
+        ).fetch("cgt-1")
+        assert result.state is TaskState.SUCCEEDED
+        assert result.progress == 100
+        assert result.asset_url.endswith("/xxx")
+        assert result.ratio == "16:9"
+        assert result.resolution == "720p"
+        assert result.billing_tokens == 108900
+        assert result.failure is None
+
+    def test_asset_url_carries_its_expiry_so_callers_cannot_treat_it_as_permanent(self):
+        """厂商 URL 只有 24 小时有效期，存成 external_url 会变死链。
+
+        把时效随 URL 一起给出，上层就没有理由把它当永久地址。
+        """
+        result = provider_with(
+            lambda r: httpx.Response(200, json=OFFICIAL_SUCCESS_PAYLOAD)
+        ).fetch("cgt-1")
+        assert result.asset_url_expires_in_sec == 24 * 3600
+
+    def test_queued_maps_to_pending(self):
+        def handler(request):
+            return httpx.Response(
+                200, json={"id": "x", "status": "queued", "content": {}}
+            )
+
+        assert provider_with(handler).fetch("x").state is TaskState.PENDING
+
+    def test_running_has_zero_progress_because_the_vendor_gives_none(self):
+        # 不是缺陷：客户端契约本就不承诺平滑进度（ADR-023 §10）。
+        def handler(request):
+            return httpx.Response(
+                200, json={"id": "x", "status": "running", "content": {}}
+            )
+
+        result = provider_with(handler).fetch("x")
+        assert result.state is TaskState.RUNNING
+        assert result.progress == 0
+
+    def test_cancelled_is_terminal_without_a_failure(self):
+        def handler(request):
+            return httpx.Response(
+                200, json={"id": "x", "status": "cancelled", "content": {}}
+            )
+
+        result = provider_with(handler).fetch("x")
+        assert result.state is TaskState.CANCELLED
+        assert result.state.is_terminal
+        assert result.failure is None
+
+    def test_expired_maps_to_failed_and_stays_retryable(self):
+        """`expired` 只见于创建接口的回调说明，查询接口文档没列。
+
+        只按查询文档实现会让超时任务掉进未知分支。它是超时而非内容问题，故可重试；
+        且厂商 expired 时不一定给 error.code，故单独分类而非走通用路径。
+        """
+
+        def handler(request):
+            return httpx.Response(
+                200, json={"id": "x", "status": "expired", "content": {}}
+            )
+
+        result = provider_with(handler).fetch("x")
+        assert result.state is TaskState.FAILED
+        assert result.failure.retry is Retry.BACKOFF
+        assert result.failure.provider_code == "TaskExpired"
+
+    def test_unknown_status_keeps_the_task_alive(self):
+        # 厂商新增状态时把在途任务判死更糟，最坏是等到我们自己的超时。
+        def handler(request):
+            return httpx.Response(
+                200, json={"id": "x", "status": "brand_new", "content": {}}
+            )
+
+        assert provider_with(handler).fetch("x").state is TaskState.RUNNING
+
+    def test_succeeded_without_a_url_is_treated_as_a_retryable_failure(self):
+        """厂商说成功却没给 URL——那不是成功。
+
+        放一个没有产物的「成功」给上层，只会把空指针推到更远的地方。
+        """
+
+        def handler(request):
+            return httpx.Response(
+                200, json={"id": "x", "status": "succeeded", "content": {}}
+            )
+
+        result = provider_with(handler).fetch("x")
+        assert result.state is TaskState.FAILED
+        assert result.failure.retry is Retry.BACKOFF
+        assert result.failure.provider_code == "MissingVideoUrl"
+
+
+class TestFetchFailureClassification:
+    def test_content_rejection_is_not_retried(self):
+        def handler(request):
+            return httpx.Response(
+                200,
+                json={
+                    "id": "x",
+                    "status": "failed",
+                    "content": {},
+                    "error": {
+                        "code": "OutputVideoSensitiveContentDetected",
+                        "message": "output may contain sensitive information",
+                    },
+                },
+            )
+
+        result = provider_with(handler).fetch("x")
+        assert result.failure.retry is Retry.NEVER
+        assert result.failure.user_facing is UserFacing.CONTENT_REJECTED
+        # 原始码与 message 都要带到上层——排查时它们是唯一线索。
+        assert result.failure.provider_code == "OutputVideoSensitiveContentDetected"
+        assert "sensitive" in result.failure.message
+
+
+class TestDurationPrecision:
+    def test_frames_over_fps_is_exact(self):
+        # 官方例子：133 帧 / 24fps = 5.54 秒。
+        def handler(request):
+            return httpx.Response(
+                200,
+                json={
+                    "id": "x",
+                    "status": "succeeded",
+                    "frames": 133,
+                    "framespersecond": 24,
+                    "content": {"video_url": "https://v"},
+                },
+            )
+
+        assert provider_with(handler).fetch("x").duration_ms == 5542
+
+    def test_falls_back_to_the_vendors_floored_seconds(self):
+        # duration 与 frames 只返回其一。只有 duration 时它是「帧数/24 向下取整」
+        # 的约数，最多可能短 1 秒——可用，但不可当精确值。
+        def handler(request):
+            return httpx.Response(
+                200,
+                json={
+                    "id": "x",
+                    "status": "succeeded",
+                    "duration": 5,
+                    "content": {"video_url": "https://v"},
+                },
+            )
+
+        assert provider_with(handler).fetch("x").duration_ms == 5000
+
+    def test_none_rather_than_a_made_up_zero(self):
+        def handler(request):
+            return httpx.Response(
+                200, json={"id": "x", "status": "running", "content": {}}
+            )
+
+        assert provider_with(handler).fetch("x").duration_ms is None
+
+
+class TestCoverImage:
+    def test_last_frame_becomes_the_cover_url(self):
+        """尾帧图可免费充当视频封面——iteration-4/03 R13 把缩略图列为视频成本项，
+        实际上厂商白给。"""
+
+        def handler(request):
+            return httpx.Response(
+                200,
+                json={
+                    "id": "x",
+                    "status": "succeeded",
+                    "content": {
+                        "video_url": "https://v",
+                        "last_frame_url": "https://f.png",
+                    },
+                },
+            )
+
+        assert provider_with(handler).fetch("x").cover_url == "https://f.png"
+
+
+class TestPixelDimensionsAreLeftBlankOnPurpose:
+    def test_no_invented_dimensions(self):
+        """火山只返回 ratio 与 resolution，不返回像素值。
+
+        换算需要一张随模型族变化的对照表；在 V7 目录种子定下模型族之前，
+        留空比编一个可能错的尺寸好。
+        """
+        result = provider_with(
+            lambda r: httpx.Response(200, json=OFFICIAL_SUCCESS_PAYLOAD)
+        ).fetch("x")
+        assert result.width_px is None
+        assert result.height_px is None
+        # 但原始信息要透出，供上层或后续实现换算。
+        assert (result.ratio, result.resolution) == ("16:9", "720p")

tests/test_assembly.py

@@ -0,0 +1,89 @@
+"""装配根与注册表的测试。"""
+
+from __future__ import annotations
+
+import pytest
+
+from app.assembly import build_registry
+from app.config import Settings, VolcengineSettings
+from app.domain.generation import (
+    GenerationRequest,
+    GenerationResult,
+    MediaKind,
+    TaskState,
+)
+from app.providers.base import GenerationProvider
+from app.providers.registry import ProviderRegistry
+from app.providers.volcengine.video import PROVIDER_CODE
+
+FAKE_API_KEY = "dummy-api-key-for-tests"
+
+
+class StubProvider(GenerationProvider):
+    def __init__(self, code: str, kind: MediaKind = MediaKind.VIDEO):
+        self._code = code
+        self._kind = kind
+
+    @property
+    def provider_code(self) -> str:
+        return self._code
+
+    @property
+    def media_kind(self) -> MediaKind:
+        return self._kind
+
+    def submit(self, request: GenerationRequest) -> str:
+        return "stub-1"
+
+    def fetch(self, provider_request_id: str) -> GenerationResult:
+        return GenerationResult(provider_request_id, TaskState.RUNNING)
+
+
+class TestRegistry:
+    def test_finds_by_code(self):
+        registry = ProviderRegistry([StubProvider("a"), StubProvider("b")])
+        assert registry.find("a").provider_code == "a"
+        assert registry.codes == frozenset({"a", "b"})
+
+    def test_unknown_code_returns_none(self):
+        """对已持久化的任务，未知 code 意味着当初提交它的适配器已被移除。
+
+        调用方应当作不可重试的失败，而不是循环重试到预算耗尽。
+        """
+        assert ProviderRegistry([StubProvider("a")]).find("gone") is None
+
+    def test_duplicate_registration_is_rejected(self):
+        # 静默覆盖会让「哪个实例在处理任务」取决于注册顺序，是最难查的一类问题。
+        with pytest.raises(ValueError, match="重复注册"):
+            ProviderRegistry([StubProvider("same"), StubProvider("same")])
+
+
+class TestAssembly:
+    def test_seedance_is_registered_when_credentials_exist(self):
+        registry = build_registry(
+            Settings(volcengine=VolcengineSettings(api_key=FAKE_API_KEY))
+        )
+        provider = registry.find(PROVIDER_CODE)
+        assert provider is not None
+        assert provider.media_kind is MediaKind.VIDEO
+
+    def test_missing_credentials_skip_the_provider_instead_of_failing_startup(self):
+        """本地只配了一家凭证时，服务仍应起得来。
+
+        真正的失败点在调用时：find() 返回 None，调用方按「适配器不可用」处理，
+        而不是拿到一个半死的实例。
+        """
+        registry = build_registry(Settings(volcengine=VolcengineSettings(api_key="")))
+        assert registry.find(PROVIDER_CODE) is None
+        assert len(registry) == 0
+
+    def test_registered_code_matches_the_catalog_column(self):
+        """注册表的键必须与 creation.generation_models.provider_code 同值。
+
+        对不上时，Java 侧按任务行的快照来查会一直查不到适配器——而症状是任务
+        永远卡在排队，离根因很远。
+        """
+        registry = build_registry(
+            Settings(volcengine=VolcengineSettings(api_key=FAKE_API_KEY))
+        )
+        assert PROVIDER_CODE in registry.codes

tests/test_config.py

@@ -0,0 +1,98 @@
+"""配置加载的测试。
+
+重点在两处：非法值必须**响亮失败而不是静默回落**（一次配置笔误变成难查的行为
+偏差是最糟的结果），以及密钥**绝不能有默认值**。
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from app.config import ServiceAuthSettings, Settings, VolcengineSettings
+
+
+class TestCredentialsHaveNoDefaults:
+    def test_api_key_defaults_to_empty_not_a_placeholder(self, monkeypatch):
+        monkeypatch.delenv("ARK_API_KEY", raising=False)
+        assert VolcengineSettings.from_env().api_key == ""
+
+    def test_service_jwt_public_key_defaults_to_empty(self, monkeypatch):
+        monkeypatch.delenv("PATBOND_SERVICE_JWT_PUBLIC_KEY", raising=False)
+        assert ServiceAuthSettings.from_env().public_key == ""
+
+    def test_missing_credentials_do_not_prevent_loading(self, monkeypatch):
+        """缺密钥时配置仍能加载。
+
+        服务要能起来并提供健康检查；缺配置在**调用点**响亮失败，而不是让整个
+        进程起不来——与 Java 侧 JwtVerifier/ServiceTokenSigner 的既有约定一致。
+        """
+        for name in ("ARK_API_KEY", "PATBOND_SERVICE_JWT_PUBLIC_KEY"):
+            monkeypatch.delenv(name, raising=False)
+        settings = Settings.from_env()
+        assert settings.volcengine.api_key == ""
+        assert settings.service_auth.public_key == ""
+
+
+class TestEnvParsing:
+    def test_values_are_trimmed(self, monkeypatch):
+        # 从 .env 或 compose 里复制粘贴时尾随空格极常见。
+        monkeypatch.setenv("ARK_API_KEY", "  dummy-value  ")
+        assert VolcengineSettings.from_env().api_key == "dummy-value"
+
+    def test_int_override(self, monkeypatch):
+        monkeypatch.setenv("ARK_TASK_EXPIRY_SEC", "7200")
+        assert VolcengineSettings.from_env().task_expiry_sec == 7200
+
+    def test_float_override(self, monkeypatch):
+        monkeypatch.setenv("ARK_TIMEOUT_SEC", "2.5")
+        assert VolcengineSettings.from_env().timeout_sec == 2.5
+
+    def test_malformed_int_fails_loudly_instead_of_falling_back(self, monkeypatch):
+        # 静默回落到默认值会让一次笔误变成难查的行为偏差。
+        monkeypatch.setenv("ARK_TASK_EXPIRY_SEC", "一小时")
+        with pytest.raises(ValueError, match="不是合法整数"):
+            VolcengineSettings.from_env()
+
+    def test_malformed_float_fails_loudly(self, monkeypatch):
+        monkeypatch.setenv("ARK_TIMEOUT_SEC", "fast")
+        with pytest.raises(ValueError, match="不是合法数字"):
+            VolcengineSettings.from_env()
+
+    def test_empty_string_uses_the_default(self, monkeypatch):
+        # compose 里 `ARK_TIMEOUT_SEC: ${FOO:-}` 展开成空串是常见情形。
+        monkeypatch.setenv("ARK_TIMEOUT_SEC", "")
+        assert VolcengineSettings.from_env().timeout_sec == 10.0
+
+
+class TestVendorConstraintsAreEnforcedAtStartup:
+    """厂商的硬性区间在启动时校验，而不是等提交时被拒。"""
+
+    def test_expiry_below_the_vendor_minimum_is_rejected(self):
+        with pytest.raises(ValueError, match=r"\[3600, 259200\]"):
+            VolcengineSettings(task_expiry_sec=60)
+
+    def test_expiry_above_the_vendor_maximum_is_rejected(self):
+        with pytest.raises(ValueError, match=r"\[3600, 259200\]"):
+            VolcengineSettings(task_expiry_sec=300000)
+
+    def test_the_boundaries_themselves_are_allowed(self):
+        assert VolcengineSettings(task_expiry_sec=3600).task_expiry_sec == 3600
+        assert VolcengineSettings(task_expiry_sec=259200).task_expiry_sec == 259200
+
+    def test_default_is_the_minimum_not_the_vendor_default(self):
+        # 厂商默认 48 小时，被放弃的任务会在厂商侧挂两天。
+        assert VolcengineSettings().task_expiry_sec == 3600
+
+
+class TestServiceAuthDefaults:
+    def test_audience_and_type_have_sensible_defaults(self, monkeypatch):
+        for name in (
+            "PATBOND_SERVICE_JWT_AUDIENCE",
+            "PATBOND_SERVICE_JWT_ISSUER",
+            "PATBOND_SERVICE_JWT_TYPE",
+        ):
+            monkeypatch.delenv(name, raising=False)
+        auth = ServiceAuthSettings.from_env()
+        assert auth.audience == "patbond-ai"
+        assert auth.issuer == "patbond-api"
+        assert auth.required_type == "service"

tests/test_layering.py

@@ -0,0 +1,127 @@
+"""分层边界的测试。
+
+这些断言不测业务，测的是**架构约束本身**。分层的价值全在「换供应商不动上层」，
+而这个性质只要有人无意加一行 import 就会悄悄失效——没有测试守着，它迟早会失效。
+"""
+
+from __future__ import annotations
+
+import ast
+import pathlib
+
+APP = pathlib.Path(__file__).resolve().parent.parent / "app"
+
+
+def imports_of(path: pathlib.Path) -> set[str]:
+    tree = ast.parse(path.read_text(encoding="utf-8"))
+    names: set[str] = set()
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            names.update(alias.name for alias in node.names)
+        elif isinstance(node, ast.ImportFrom) and node.module:
+            names.add(node.module)
+    return names
+
+
+def modules_under(*parts: str) -> list[pathlib.Path]:
+    return sorted((APP.joinpath(*parts)).rglob("*.py"))
+
+
+class TestDomainIsIndependent:
+    """领域层是「换供应商不动上层」的支点。它一旦依赖厂商或 HTTP，分层就名存实亡。"""
+
+    def test_domain_imports_no_provider_code(self):
+        for module in modules_under("domain"):
+            offenders = {name for name in imports_of(module) if "providers" in name}
+            assert not offenders, f"{module.name} 不应依赖服务商代码：{offenders}"
+
+    def test_domain_imports_no_http_library(self):
+        for module in modules_under("domain"):
+            offenders = {
+                name
+                for name in imports_of(module)
+                if name.split(".")[0] in {"httpx", "requests"}
+            }
+            assert not offenders, f"{module.name} 不应依赖 HTTP 库：{offenders}"
+
+    def test_domain_does_not_read_configuration(self):
+        # 领域对象应当是纯数据；让它自己读配置会使单测必须摆弄环境变量。
+        for module in modules_under("domain"):
+            offenders = {
+                name for name in imports_of(module) if name in {"os", "app.config"}
+            }
+            assert not offenders, f"{module.name} 不应读取配置：{offenders}"
+
+
+class TestAbstractionDoesNotLeak:
+    def test_provider_base_knows_no_specific_vendor(self):
+        """抽象接口若 import 了火山，新增第二家供应商就得改抽象层。"""
+        offenders = {
+            name
+            for name in imports_of(APP / "providers" / "base.py")
+            if "volcengine" in name
+        }
+        assert not offenders, f"抽象层不应依赖具体服务商：{offenders}"
+
+    def test_registry_knows_no_specific_vendor(self):
+        offenders = {
+            name
+            for name in imports_of(APP / "providers" / "registry.py")
+            if "volcengine" in name
+        }
+        assert not offenders, f"注册表不应依赖具体服务商：{offenders}"
+
+
+class TestTransportStaysBusinessAgnostic:
+    def test_ark_client_does_not_import_video_business(self):
+        """传输层复用的前提是它不认识「视频」。
+
+        一旦 client.py 依赖了 video.py，接火山的图片接口时就没法复用它，
+        只能再抄一遍鉴权与错误解包——那正是拆分要避免的。
+        """
+        offenders = {
+            name
+            for name in imports_of(APP / "providers" / "volcengine" / "client.py")
+            if name.endswith((".video", ".image", "models"))
+        }
+        assert not offenders, f"传输层不应依赖具体业务模块：{offenders}"
+
+
+class TestVendorVocabularyIsContained:
+    def test_only_the_volcengine_package_mentions_ark_paths(self):
+        """厂商特有的路径、字段名只应出现在其适配包内。
+
+        `framespersecond` 这种厂商拼法漏到上层，就意味着上层已经知道自己在跟
+        火山说话——换供应商时会牵连到它。
+        """
+        vendor_markers = ("framespersecond", "contents/generations", "ark.cn-beijing")
+        for module in sorted(APP.rglob("*.py")):
+            if "volcengine" in module.parts:
+                continue
+            text = module.read_text(encoding="utf-8")
+            for marker in vendor_markers:
+                assert marker not in text, (
+                    f"{module.relative_to(APP)} 出现了厂商专有词汇 {marker!r}，"
+                    f"应收拢到 providers/volcengine/ 内"
+                )
+
+
+class TestDependencyDirectionIsOneWay:
+    """配置 → 适配层，绝不反向。
+
+    `config.py` 作为装配根知道有哪家供应商是合理的；反过来，适配层一旦自己去读
+    配置，依赖注入就失效了——单测又得摆弄环境变量，而且同一进程内没法跑两套
+    不同配置的适配器实例。
+    """
+
+    def test_providers_do_not_read_config(self):
+        for module in modules_under("providers"):
+            offenders = {
+                name
+                for name in imports_of(module)
+                if name == "app.config" or name == "os"
+            }
+            assert not offenders, (
+                f"{module.relative_to(APP)} 不应自行读取配置或环境变量："
+                f"{offenders}。请通过构造参数注入。"
+            )