T15 · 报告域骨架——触发/配额/异步链路/中间态恢复

对应任务：T-P3-07（触发与频控、配额生效）、T-P3-09 🔴（失败不需重新上传照片——中间态落库+分步恢复）、架构 §7.1 步骤 1~~7 与 22~~24 ｜分支：task/t15-report-skeleton ｜ worktree：/home/xsl/meida_backend-t15 先读《README.md（协作规程）》，遵守全部纪律。第四批三个任务（T15/T16/T17）拼成 M3 报告流水线：你是骨架与编排，T16 供视觉/诊断阶段，T17 供文案/校验阶段——你们互不改对方文件，靠协调提交定好的阶段注册机制衔接。

1. 目标

报告域的编排骨架：POST /reports（资格校验→终身配额预扣→建 tk_task→入 cpu 队列）→ worker 执行可注入的流水线（阶段动态装配、中间态逐段落库、视觉结果可复用不重跑）→ 失败自动退款；GET /reports、GET /reports/{id}、POST /reports/{id}/regenerate（走 report_regen 月配额）。

2. 已就绪的契约（协调提交已提供，禁改 models/migrations/config/errors）

表（迁移 `0006`，ORM `RpReport`/`RpInput`/`RpShare`，先读 models.py 核对）

rp_report：report_id PK、user_id、tier('basic'/'full')、status('pending'/'running'/'succeeded'/'failed')、version、result JSONB（流水线中间态与最终产物都写这里，按 stage 键增量更新）、model_info JSONB、cost_cents Numeric、created_at/completed_at；Index(user_id,created_at)
rp_input：input_id PK、report_id、questionnaire JSONB、photo_meta JSONB、derived_features JSONB（视觉中间态——T16 的 vision 阶段写，已存在则跳过视觉＝T-P3-09 的核心）；UNIQUE(report_id)
rp_share：share_id PK、report_id、share_token UNIQUE、expire_at、view_count、revoked（V1 只建结构，路由不做）

错误码：`REPORT_NOT_FOUND = 40420`（404）、`REPORT_ALREADY_EXISTS = 40920`（409「本期报告已生成过」，hint「每 30 天可重新生成一次」——文案不含价格字样）

配置：`report_pipeline_stages: list[str]`（dotted path 列表，默认 `["packages.report.vision_stage:VisionStage","packages.report.diagnosis_stage:DiagnosisStage","packages.report.copy_stage:CopyStage","packages.report.validation_stage:ValidationStage"]`）、`report_vision_provider: str = "deterministic"`、`report_tier: str = "basic"`

阶段注册机制（协调提交的 `packages/report/pipeline.py` 骨架里有接口约定，你实现）

Stage Protocol：name: str + async def run(ctx: ReportContext) -> None（阶段把产物写进 ctx.report.result[self.name]，异常即流水线失败）；
ReportContext（frozen dataclass）：report: RpReport、rp_input: RpInput、session、settings、gateway: LLMGateway | None（worker 装配时注入——T17 的阶段要用）；
assemble_stages(settings) -> list[Stage]：用 importlib 按 dotted path 动态导入（这是刻意的插件式装配：T16/T17 各自建阶段模块，不碰你的文件）。模块/类不存在时抛 AppError(CONFIG_ERROR) 并列出缺失项——绝不静默跳过（静默=流水线空跑还扣配额）。动态导入绕过 import-linter 静态检查，注释写明这一点与理由。

复用设施（先读再用）

BillingService.reserve/commit/refund（quota_key：首次 report_basic（lifetime）、重新生成 report_regen（month））；幂等键沿用 reserve_idempotency_key 风格；
TaskService.submit(...)（幂等键=request_key）+ tk_task 状态机；celery task_routes 已预设 meida.report.*→cpu；
参考现成范式：apps/api/routers/tasks.py 的 ping 投递 + apps/worker/tasks.py 的 _run_ping（asyncio.run 包装）。

3. 实现要点

允许改动：packages/report/**（pipeline.py、service.py、repo.py、schemas.py、__init__.py、README.md）、apps/api/routers/reports.py（填充协调提交的空桩）、apps/worker/report_tasks.py（新建）、测试文件。vision_stage.py / diagnosis_stage.py / copy_stage.py / validation_stage.py 四个文件归 T16/T17，禁建禁改——你的流水线只认配置里的 dotted path。

pipeline.py：Stage Protocol、ReportContext、assemble_stages（动态导入+缺失即 CONFIG_ERROR）、run_pipeline(ctx, stages)——逐阶段执行，每阶段完成后 ctx.report.result[stage.name] = {...} 并 flush 落库（中间态持久化）；阶段异常：tk_task→failed + refund + rp_report.status='failed' + error 写 result，异常上抛给 celery 重试逻辑前先保证退款事务提交（参考 agent service 的事务纪律）。
service.py ReportService：
- create_report(user_id, questionnaire, photo_object_key, request_key)：查该用户已 succeeded 的 basic 报告 → 有则 40920（首次终身 1 次的资格语义，FR 定义见需求文档 §2.1）；reserve(report_basic, amount=1) → 建 rp_report(pending)+rp_input → tk_task(idempotency_key=request_key) → 入队 → 返回 task_id+report_id。reserve 与 tk_task 同事务，入队在事务提交后（队列消息不能早于数据落库）；
- regenerate(report_id, request_key)：目标报告须 succeeded；reserve(report_regen) → 复用 rp_input（derived_features 保留——这就是"失败/重生成不需重新上传"）→ version+1 新报告行；
- get_report / list_reports（归属校验，result 透出但 failed 只透错误摘要）；
- 幂等：同 request_key 二次提交返回同一 task_id，不重复预扣（tk_task 幂等键 + reserve 幂等键双重保证）。
worker/report_tasks.py：@celery_app.task(name="meida.report.generate", bind=True, max_retries=1)——开 session、装配 gateway（worker 侧 build_llm_gateway(settings)——deps.py 是 apps.api 的，worker 里自建装配函数，不 import apps.api）、assemble_stages、执行 run_pipeline、成功后 commit 配额 + status=succeeded + completed_at + 成本聚合写 model_info。重试语义：重试时若 rp_input.derived_features 已存在，视觉阶段自然跳过（T16 实现），网络类异常才重试。
路由 reports.py（prefix /reports）：POST /（body：questionnaire dict + photo_object_key + 可选 request_key）、GET /、GET /{id}、POST /{id}/regenerate。响应模型放 packages/report/schemas.py。
测试钩子：集成测试不启真 celery——直接调 run_pipeline（注入 Fake 阶段）+ 独立测 celery 任务函数（task.run() 直调）。先看 tests/integration/test_task_bonus.py 里已有的 dev-token 与 app fixture 模式。

4. 测试要求

单测 tests/unit/test_report_pipeline.py：assemble_stages 缺失模块 → CONFIG_ERROR 且列出缺失项；run_pipeline 两阶段顺序执行、产物按键写入、第二阶段抛错 → 状态 failed + 已完成阶段产物仍在。
集成 tests/integration/test_report_flow.py（新文件，Fake 阶段注入——把配置里的 stages 用 monkeypatch 换成文件内定义的两个 Fake 阶段）：
1. dev-token → POST /reports → 200，task/report id；tk_task PENDING、bl_ledger reserved 一行（report_basic lifetime）；
2. 直调 worker 函数（Fake 阶段成功）→ rp_report succeeded、result 含两个阶段键、ledger committed；
3. 重复同 request_key → 同 task_id、无第二笔 reserve；
4. 首次资格：succeeded 后再 POST /reports → 40920；
5. 失败退款：Fake 阶段二抛错 → status failed、ledger 出现 refund、rp_input.derived_features 保留（Fake 阶段一写的）；
6. regenerate：走 report_regen 月配额、version+1、复用 derived_features。

5. 禁改清单

通用禁改之外：packages/report/{vision,vision_stage,diagnosis_stage,copy,copy_stage,validation,validation_stage,dimensions}.py（T16/T17 所有）、全部已有测试、apps/worker/tasks.py。

6. 验收标准（T-P3-07/09）

首次终身资格校验与 40920；regenerate 走月配额；
幂等提交不重复扣减；
阶段产物逐段落库；失败自动退款且视觉中间态保留（重试/重生成不重跑视觉的结构保证）；
全部检查绿（mypy 范围：T6 同口径 + packages/report）。

7. 提交信息

feat(report): orchestration skeleton with staged pipeline and refund-on-failure

POST /reports 骨架：资格校验→终身配额预扣→tk_task→cpu 队列；动态装配
流水线（dotted-path 注册、缺阶段即 CONFIG_ERROR）；中间态逐段落库、失败
退款、derived_features 跨重试/重生成复用（T-P3-07/09）。

偏差说明：<如有写在这里，无则写"无">

8.7 KiB Raw Blame History Unescape Escape