feat(uploads): 本地磁盘直传（COS 降级路线）+ 报告照片存在性校验

- POST /uploads/direct：原始字节 body + Content-Type 申报 + scene query（默认 report_photo），场景/类型/大小三重校验复用 storage 既有规则，落盘 var/uploads/<scene>/<user_id>/<uuid>.<ext>（config 新增 upload_dir），key 布局与 COS 同构；不加 python-multipart 依赖
- storage/local.py：save_direct_upload + require_local_upload（resolve 归并 + is_relative_to 归属收敛 + 存在性校验，越界/模拟/不存在一律 404）
- report/service.create_report：资格校验后、配额预扣前插入照片存在性校验（模拟 key 被拦且不消耗额度；regenerate 复用已验 photo_meta 不复核）
- 测试：test_upload_direct 14 单测 + test_upload_direct_flow 4 集成（真 key 全链路/假 key 404 零流水/401/40010）

偏差说明：test_report_flow._create_report 改走 /uploads/direct 拿真 key（存在性校验必然封死假 key），40920 与幂等重放两条断言语义未动。

.env.example

@@ -91,9 +91,11 @@ LLM_BUDGET_TIERS_JSON=[{"threshold":0.6,"name":"warn","blocked":[]},{"threshold"
 REFRESH_TOKEN_TTL_SECONDS=2592000
 
 # ── 任务奖励（FR-BILL-03/04）───────────────────────────────────────────────
-# amount 均为占位值 1：正式数值待产品在产品方案 §3.2 定稿后改此处，不改代码。
+# 产品方案 §3.2 定稿目录（2026-09-11）：签到+10 对话/日、反馈+2 发型/日、
+# 分享+3 换装/日、首次速测+1 重生成/终身。invite_friend/profile_complete 下架
+# （无可信完成信号，不下发即下架，claim 返回 40410）。
 # claim_period 决定幂等粒度（同一任务同周期只能领一次）。
-TASK_BONUS_RULES_JSON={"sign_in":{"quota_key":"agent_chat","amount":1,"claim_period":"day"},"profile_complete":{"quota_key":"agent_chat","amount":1,"claim_period":"lifetime"},"report_feedback":{"quota_key":"report_regen","amount":1,"claim_period":"day"},"share_report":{"quota_key":"tryon","amount":1,"claim_period":"day"},"invite_friend":{"quota_key":"tryon","amount":1,"claim_period":"day"}}
+TASK_BONUS_RULES_JSON={"sign_in":{"quota_key":"agent_chat","amount":10,"claim_period":"day"},"report_feedback":{"quota_key":"hairstyle","amount":2,"claim_period":"day"},"share_report":{"quota_key":"tryon","amount":3,"claim_period":"day"},"first_report":{"quota_key":"report_regen","amount":1,"claim_period":"lifetime"}}
 
 # ── 客户端埋点（FR-SYS-06）─────────────────────────────────────────────────
 EVENTS_MAX_BATCH=50

apps/api/routers/quota.py

@@ -71,8 +71,8 @@ async def claim_task_bonus(
 ) -> dict[str, object]:
     """同一任务同一周期只能领一次（重复领返回 40910），受规则封顶（40310）.
 
-    奖励数值由 ``TASK_BONUS_RULES_JSON`` 配置驱动（FR-BILL-04），当前是占位值，
-    产品定稿后改 env 即生效。
+    奖励数值由 ``TASK_BONUS_RULES_JSON`` 配置驱动（FR-BILL-04），2026-09-11 起默认值
+    即产品定稿（产品方案 §3.2）；report_feedback/first_report 有服务端凭据校验（40311）。
     """
     claim = await billing.claim_task_bonus(
         user_id=user_id,

apps/api/routers/uploads.py

@@ -4,14 +4,20 @@
 字节流由客户端直传 COS。真实联调需在 `.env` 配 COS 四件套
 （COS_REGION / COS_SECRET_ID / COS_SECRET_KEY + 场景桶配置
 UPLOAD_SCENES_JSON）；本地单测注入 FakeSigner，不依赖任何真实 COS。
+
+``POST /uploads/direct`` — 本地磁盘直传（无 COS 四件套的降级路线，docs
+提交 9228f67 决策）：body 即图片原始字节，Content-Type 头申报类型，
+scene 走 query（默认 report_photo）。校验口径与 presign 完全一致，
+落盘 ``settings.upload_dir``，返回的 object_key 供建报告等接口回传。
 """
 
 from __future__ import annotations
 
-from fastapi import APIRouter
+from fastapi import APIRouter, Query, Request
 
-from apps.api.deps import CurrentUserId, StorageServiceDep
+from apps.api.deps import CurrentUserId, SettingsDep, StorageServiceDep
 from packages.shared.response import ok
+from packages.storage.local import save_direct_upload
 from packages.storage.schemas import PresignRequest
 
 router = APIRouter(prefix="/uploads", tags=["uploads"])
@@ -25,3 +31,19 @@ async def presign_upload(
 ) -> dict[str, object]:
     presigned = storage.build_presign(user_id=user_id, request=request)
     return ok(presigned.model_dump())
+
+
+@router.post("/direct", summary="本地磁盘直传图片（无 COS 降级路线）：body 为原始字节")
+async def direct_upload(
+    user_id: CurrentUserId,
+    settings: SettingsDep,
+    request: Request,
+    scene: str = Query(default="report_photo", min_length=1, max_length=64),
+) -> dict[str, object]:
+    data = await request.body()
+    # Content-Type 头可能带参数（如 "image/jpeg; boundary=…"），只取主类型。
+    content_type = request.headers.get("content-type", "").split(";")[0].strip().lower()
+    result = save_direct_upload(
+        settings, user_id=user_id, scene=scene, content_type=content_type, data=data
+    )
+    return ok(result.model_dump())

packages/billing/repo.py

@@ -6,11 +6,13 @@
 
 from __future__ import annotations
 
+from datetime import datetime
+
 from sqlalchemy import select
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from packages.shared.models import Ledger, Quota, QuotaRule
+from packages.shared.models import Ledger, Quota, QuotaRule, RpFeedback, RpReport
 
 
 async def get_rule(session: AsyncSession, quota_key: str) -> QuotaRule | None:
@@ -77,3 +79,40 @@ async def list_ledger(session: AsyncSession, user_id: int, limit: int = 50) -> l
 def add_ledger(session: AsyncSession, ledger: Ledger) -> Ledger:
     session.add(ledger)
     return ledger
+
+
+# -- 任务完成凭据探测（T2b，rp_* 只读） --------------------------------------
+# 分层契约（pyproject import-linter layers）禁止 packages.billing 依赖
+# packages.report，而 claim 完成校验又必须读 rp_feedback/rp_report 的服务端
+# 可信信号 —— 直接查 shared.models 的只读 exists 探测是唯一合规通道。
+# 纪律：billing 对 rp_* 永远只读，绝不写入；「完成判定」语义仍归 report 域。
+
+#: rp_report.status 的「已完成」取值：与 packages/report/repo.py 的
+#: STATUS_SUCCEEDED 对齐（同上，分层契约不允许 import 该常量，故复制字面量）。
+_REPORT_STATUS_SUCCEEDED = "succeeded"
+
+
+async def has_feedback_between(
+    session: AsyncSession, *, user_id: int, start: datetime, end: datetime
+) -> bool:
+    """``[start, end)`` 内该用户是否存在反馈行（report_feedback 完成凭据）。"""
+    result = await session.execute(
+        select(RpFeedback.feedback_id)
+        .where(
+            RpFeedback.user_id == user_id,
+            RpFeedback.created_at >= start,
+            RpFeedback.created_at < end,
+        )
+        .limit(1)
+    )
+    return result.first() is not None
+
+
+async def has_completed_report(session: AsyncSession, *, user_id: int) -> bool:
+    """该用户是否存在已完成报告（first_report 完成凭据）。"""
+    result = await session.execute(
+        select(RpReport.report_id)
+        .where(RpReport.user_id == user_id, RpReport.status == _REPORT_STATUS_SUCCEEDED)
+        .limit(1)
+    )
+    return result.first() is not None

packages/billing/service.py

@@ -10,7 +10,7 @@ Nothing here decides *how much* a feature costs — that comes from
 from __future__ import annotations
 
 from dataclasses import dataclass
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta, timezone
 from typing import Any, cast
 
 from sqlalchemy.exc import IntegrityError
@@ -310,20 +310,26 @@ class BillingService:
         task_key: str,
         rules: dict[str, dict[str, Any]],
     ) -> TaskBonusClaim:
-        """领取任务奖励：发放即到账、幂等、封顶（FR-BILL-03）.
+        """领取任务奖励：完成校验 → 发放即到账、幂等、封顶（FR-BILL-03）.
 
         ``rules`` 来自 ``settings.task_bonus_rules``（FR-BILL-04 配置化）：
-        amount 目前是占位值，产品在产品方案 §3.2 定稿后改 env 即生效，
-        不需要改代码。
+        目录与数值以产品方案 §3.2 定稿为准，改 env 即生效，不需要改代码。
         """
         conf = rules.get(task_key)
         if conf is None:
             raise AppError(ErrorCode.TASK_BONUS_UNKNOWN, data={"task_key": task_key})
         quota_key, amount, claim_period = _parse_task_rule(task_key, conf)
 
+        now = datetime.now(UTC)
+        # 完成凭据校验（T2b）必须在幂等快捷路径之前：无凭据的 claim 一律 40311，
+        # 不得因为「本周期已领过」之类的既有状态绕过校验语义（顺序反过来时
+        # 已领用户会看到 40910 而非 40311，两者都可接受，但先校验口径更一致）。
+        await self._require_task_eligibility(
+            user_id=user_id, task_key=task_key, quota_key=quota_key, now=now
+        )
+
         rule = await self._require_rule(quota_key)
         quota_period_type = cast(quota_rules.PeriodType, rule.period_type)
-        now = datetime.now(UTC)
         # 幂等粒度跟任务的 claim_period 走，额度行周期跟 bl_quota_rule 走：
         # 「每天可领」的任务可以给「按月计量」的额度加成，两套周期不能混。
         quota_pk = quota_rules.period_key(now, quota_period_type, self._tz_offset)
@@ -416,6 +422,38 @@ class BillingService:
 
     # -- helpers ------------------------------------------------------------
 
+    async def _require_task_eligibility(
+        self, *, user_id: int, task_key: str, quota_key: str, now: datetime
+    ) -> None:
+        """claim 前的完成凭据校验（T2b / 产品方案 §3.2）：无凭据抛 40311.
+
+        映射写死、不做配置化——校验逻辑与任务语义绑定，不是运营参数：
+        - ``report_feedback``：当日（UTC+8，与 ``period_key`` 同时区口径）存在
+          ``rp_feedback`` 行；
+        - ``first_report``：存在已完成（succeeded）的 ``rp_report`` 行；
+        - ``sign_in`` / ``share_report`` 不校验：签到点击行为本身即完成；分享
+          内测期端侧自报、由 ``hard_cap`` 兜底（§3.2 决策理由 4）。
+        埋点只作观测、不作发放凭据（可伪造且有节流延迟），故这里只读业务表。
+        拒绝路径不写任何流水（校验在写路径之前）。
+        """
+        eligible = True
+        if task_key == "report_feedback":
+            tz = timezone(timedelta(hours=self._tz_offset))
+            day_start = now.astimezone(tz).replace(hour=0, minute=0, second=0, microsecond=0)
+            day_start_utc = day_start.astimezone(UTC)
+            eligible = await repo.has_feedback_between(
+                self._session,
+                user_id=user_id,
+                start=day_start_utc,
+                end=day_start_utc + timedelta(days=1),
+            )
+        elif task_key == "first_report":
+            eligible = await repo.has_completed_report(self._session, user_id=user_id)
+        if not eligible:
+            QUOTA_EVENTS.labels(quota_key, "grant", "not_eligible").inc()
+            logger.info("task_bonus_not_eligible", user_id=user_id, task_key=task_key)
+            raise AppError(ErrorCode.TASK_BONUS_NOT_ELIGIBLE, data={"task_key": task_key})
+
     async def _require_rule(self, quota_key: str) -> QuotaRule:
         rule = await repo.get_rule(self._session, quota_key)
         if rule is None:

packages/report/schemas.py

@@ -17,7 +17,12 @@ class ReportCreateRequest(BaseModel):
         default_factory=dict, description="问卷答案；字段结构由产品定稿后在路由层收紧"
     )
     photo_object_key: str = Field(
-        min_length=1, max_length=255, description="临时桶里已上传照片的对象键（report_photo 场景）"
+        min_length=1,
+        max_length=255,
+        description=(
+            "已上传照片的对象键（report_photo 场景）；本地降级路线下为 "
+            "POST /uploads/direct 返回的相对路径，建报告时做存在性校验"
+        ),
     )
     request_key: str | None = Field(
         default=None, max_length=128, description="幂等键；相同键返回同一任务且不重复扣配额"

packages/report/service.py

@@ -22,6 +22,7 @@ from packages.shared.config import Settings
 from packages.shared.errors import AppError, ErrorCode, ForbiddenError
 from packages.shared.logging import get_logger
 from packages.shared.models import RpInput, RpReport, Task
+from packages.storage.local import require_local_upload
 from packages.task import repo as task_repo
 from packages.task.service import TaskService
 
@@ -101,6 +102,14 @@ class ReportService:
         if await repo.find_succeeded_basic(self._session, user_id=user_id) is not None:
             raise AppError(ErrorCode.REPORT_ALREADY_EXISTS)
 
+        # 照片存在性校验（本地磁盘降级路线）：key 必须落在
+        # upload_dir/report_photo/<本人>/ 下且文件真实存在——模拟 key 与越界
+        # key 一律 404。校验在配额预扣之前，不合格的照片不消耗额度。
+        # regenerate 不复核：它复用的是上一版已验过的 photo_meta（T-P3-09）。
+        require_local_upload(
+            self._settings, user_id=user_id, scene="report_photo", object_key=photo_object_key
+        )
+
         report = RpReport(
             report_id=next_id(),
             user_id=user_id,

packages/shared/config.py

@@ -131,19 +131,15 @@ class Settings(BaseSettings):
     agent_history_turns: int = 10
 
     # --- 任务奖励（FR-BILL-03/04，配置化）-----------------------------------
-    # amount 全部是占位值 1：正式数值待产品在产品方案 §3.2 定稿后改 env，不改代码。
+    # 默认值与产品方案 §3.2 定稿目录一致（2026-09-11）：4 项上线任务；
+    # invite_friend/profile_complete 无可信完成信号，不下发即下架（claim 40410）。
     # claim_period 决定幂等粒度（同一任务同周期只能领一次）。
     task_bonus_rules: dict[str, dict[str, Any]] = Field(
         default_factory=lambda: {
-            "sign_in": {"quota_key": "agent_chat", "amount": 1, "claim_period": "day"},
-            "profile_complete": {
-                "quota_key": "agent_chat",
-                "amount": 1,
-                "claim_period": "lifetime",
-            },
-            "report_feedback": {"quota_key": "report_regen", "amount": 1, "claim_period": "day"},
-            "share_report": {"quota_key": "tryon", "amount": 1, "claim_period": "day"},
-            "invite_friend": {"quota_key": "tryon", "amount": 1, "claim_period": "day"},
+            "sign_in": {"quota_key": "agent_chat", "amount": 10, "claim_period": "day"},
+            "report_feedback": {"quota_key": "hairstyle", "amount": 2, "claim_period": "day"},
+            "share_report": {"quota_key": "tryon", "amount": 3, "claim_period": "day"},
+            "first_report": {"quota_key": "report_regen", "amount": 1, "claim_period": "lifetime"},
         },
         alias="TASK_BONUS_RULES_JSON",
     )
@@ -163,6 +159,10 @@ class Settings(BaseSettings):
     cos_presign_expire_seconds: int = 600
     upload_max_bytes: int = 20_971_520  # 20MB
     upload_allowed_types: str = "image/jpeg,image/png,image/webp"
+    # 本地磁盘直传落盘目录（无 COS 四件套的降级路线，docs 9228f67 决策）：
+    # POST /uploads/direct 把字节写到 {upload_dir}/{scene}/{user_id}/{uuid}.{ext}，
+    # object_key 为相对此目录的 POSIX 路径。相对进程 cwd，与 report_export_dir 同约定。
+    upload_dir: str = "var/uploads"
     # scene → {"bucket": "temp"|"user"|"kb"|"report"}。新增上传场景只改这里。
     upload_scenes: dict[str, dict[str, Any]] = Field(
         default_factory=lambda: {

packages/shared/errors.py

@@ -39,6 +39,7 @@ class ErrorCode(IntEnum):
     CONSENT_REQUIRED = 40302
     MINOR_RESTRICTED = 40303
     TASK_BONUS_CAP_REACHED = 40310
+    TASK_BONUS_NOT_ELIGIBLE = 40311
     NOT_FOUND = 40400
     TASK_BONUS_UNKNOWN = 40410
     REPORT_NOT_FOUND = 40420
@@ -101,6 +102,7 @@ ERROR_META: dict[ErrorCode, ErrorMeta] = {
     ErrorCode.TASK_BONUS_CAP_REACHED: ErrorMeta(
         403, "任务奖励已达上限", "明天再来，或等额度自然恢复"
     ),
+    ErrorCode.TASK_BONUS_NOT_ELIGIBLE: ErrorMeta(403, "任务尚未完成", "完成后再来领取"),
     ErrorCode.NOT_FOUND: ErrorMeta(404, "内容不存在或已删除"),
     ErrorCode.TASK_BONUS_UNKNOWN: ErrorMeta(404, "任务不存在或暂未开放"),
     ErrorCode.REPORT_NOT_FOUND: ErrorMeta(404, "报告不存在或已删除"),

packages/storage/local.py

@@ -0,0 +1,94 @@
+"""本地磁盘直传 —— 无 COS 四件套时的降级路线（docs 提交 9228f67 决策）。
+
+与 presign 共用场景/类型/大小三重校验口径（同 40010 错误码），但字节流经
+应用服务器落盘 ``settings.upload_dir``；object_key 为相对该目录的 POSIX 路径
+``{scene}/{user_id}/{uuid4hex}.{ext}``，与 COS 布局同构——后续切回 COS 直传
+只换写入实现，key 语义与下游（报告链路的 photo_object_key）不变。
+
+依赖纪律：仓库未引入 python-multipart（且不加新依赖），服务端不收
+multipart 表单；路由层以「body 即图片字节、Content-Type 头申报类型、
+scene 走 query」的原始字节口径对接本模块。
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from uuid import uuid4
+
+from pydantic import BaseModel, Field
+
+from packages.shared.config import Settings
+from packages.shared.errors import AppError, ErrorCode, NotFoundError
+from packages.shared.logging import get_logger
+from packages.storage.service import _CONTENT_TYPE_EXT, _parse_allowed_types
+
+logger = get_logger(__name__)
+
+
+class DirectUploadResponse(BaseModel):
+    """本地直传结果：object_key 交给后续业务接口（如建报告）原样回传。"""
+
+    object_key: str = Field(description="相对 upload_dir 的对象键（scene/user_id/uuid.ext）")
+    size_bytes: int = Field(description="实际落盘字节数（服务端实测，非客户端自报）")
+    content_type: str = Field(description="回显申报的 Content-Type")
+
+
+def save_direct_upload(
+    settings: Settings,
+    *,
+    user_id: int,
+    scene: str,
+    content_type: str,
+    data: bytes,
+) -> DirectUploadResponse:
+    """三重校验（场景/类型/大小）通过后落盘，与 presign 同码拒绝（40010）。
+
+    presign 只能信客户端自报的 size_bytes；直传拿到的是真实字节数，
+    空文件与超限在此一并拒绝。
+    """
+    if scene not in settings.upload_scenes:
+        raise AppError(ErrorCode.UPLOAD_REJECTED, message="不支持的上传场景")
+    if content_type not in _parse_allowed_types(settings.upload_allowed_types):
+        raise AppError(ErrorCode.UPLOAD_REJECTED)
+    ext = _CONTENT_TYPE_EXT.get(content_type)
+    if ext is None:
+        raise AppError(ErrorCode.UPLOAD_REJECTED)
+    if not data or len(data) > settings.upload_max_bytes:
+        raise AppError(ErrorCode.UPLOAD_REJECTED)
+
+    # 文件名由服务端生成（uuid 防猜），object_key 不含任何客户端输入，
+    # 落盘路径因此天然无穿越面。
+    object_key = f"{scene}/{user_id}/{uuid4().hex}.{ext}"
+    path = Path(settings.upload_dir).resolve() / object_key
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_bytes(data)
+    logger.info(
+        "direct_upload_saved",
+        scene=scene,
+        user_id=user_id,
+        object_key=object_key,
+        size_bytes=len(data),
+    )
+    return DirectUploadResponse(
+        object_key=object_key, size_bytes=len(data), content_type=content_type
+    )
+
+
+def require_local_upload(
+    settings: Settings,
+    *,
+    user_id: int,
+    scene: str,
+    object_key: str,
+) -> Path:
+    """把 object_key 收敛到 upload_dir 内并校验存在性，挡模拟/越界 key。
+
+    「不存在」与「不是本人的」一律同报 404，不向调用方泄露目录结构；
+    resolve() 先归并 ``..`` 与绝对路径，再用 is_relative_to 判定归属目录。
+    """
+    base = Path(settings.upload_dir).resolve()
+    owner_dir = base / scene / str(user_id)
+    target = (base / object_key).resolve()
+    if not target.is_relative_to(owner_dir) or not target.is_file():
+        raise NotFoundError("照片不存在或已过期，请重新上传")
+    return target

prompts/agent_chat/v1.txt

@@ -12,7 +12,8 @@
 - 用简体中文，语气平实、具体、可执行。
 - 优先给出可直接照做的搭配组合，而不是抽象的风格理论。
 - 涉及专业术语时补一句白话解释。
-- 单次回答控制在三百字以内。
+- 单次回答控制在一百五十字以内，最多 3 个要点，每点一句话说完。
+- 不要使用 Markdown 标记（**、#、- 列表符等）和表情符号，直接写口语化短句——回答会被语音朗读。
 - 当信息不足以给出建议时，先说明还需要知道什么，再给一个通用的稳妥方案。
 
 安全约束：

tests/integration/test_report_flow.py

@@ -133,6 +133,20 @@ async def _login(client: Any) -> tuple[dict[str, str], int]:
     return {"Authorization": f"Bearer {data['access_token']}"}, int(data["user_id"])
 
 
+async def _upload_photo(client: Any, headers: dict[str, str]) -> str:
+    """直传一张假 JPEG 拿真实 object_key（建报告对 photo_object_key 做存在性校验）。
+
+    字节内容不需要是真图片：上传链路只校验类型/大小，不嗅探内容。
+    """
+    resp = await client.post(
+        "/api/v1/uploads/direct?scene=report_photo",
+        headers={**headers, "Content-Type": "image/jpeg"},
+        content=b"\xff\xd8\xff\xe0" + b"\x00" * 64,
+    )
+    assert resp.status_code == 200
+    return str(resp.json()["data"]["object_key"])
+
+
 async def _create_report(
     client: Any, headers: dict[str, str], *, request_key: str | None = None
 ) -> dict[str, Any]:
@@ -141,7 +155,7 @@ async def _create_report(
         headers=headers,
         json={
             "questionnaire": {"style": "通勤", "height_cm": 168},
-            "photo_object_key": "report_photo/u1.jpg",
+            "photo_object_key": await _upload_photo(client, headers),
             "request_key": request_key or f"rk-{uuid.uuid4().hex}",
         },
     )
@@ -223,7 +237,8 @@ async def test_create_reserves_lifetime_quota_and_creates_pending_task(
     report = await _fetch_report(report_id)
     assert report.status == "pending"
     rp_input = await _fetch_input(report_id)
-    assert rp_input.photo_meta == {"object_key": "report_photo/u1.jpg"}
+    # photo_object_key 已由直传接口真实落盘（布局 report_photo/{user_id}/{uuid}.jpg）
+    assert str(rp_input.photo_meta["object_key"]).startswith(f"report_photo/{user_id}/")
     assert rp_input.derived_features == {}
 
     rows = await _ledger_rows(user_id, "report_basic")

tests/integration/test_task_bonus.py

@@ -86,15 +86,15 @@ async def test_claim_grants_immediately_and_repeats_are_rejected(client: Any) ->
     assert body["data"] == {
         "task_key": "sign_in",
         "quota_key": "agent_chat",
-        "amount": 1,  # 占位值；正式数值由产品定稿后改 TASK_BONUS_RULES_JSON
-        "bonus_granted": 1,
-        "remaining": 11,  # base_limit 10 + bonus 1 - used 0
+        "amount": 10,  # §3.2 定稿值（T2b）；改 TASK_BONUS_RULES_JSON 即生效
+        "bonus_granted": 10,
+        "remaining": 20,  # base_limit 10 + bonus 10 - used 0
     }
 
     # 发放即到账：额度视图立刻可见。
     view = await _quota_view(client, headers, "agent_chat")
-    assert view["bonus_granted"] == 1
-    assert view["remaining"] == 11
+    assert view["bonus_granted"] == 10
+    assert view["remaining"] == 20
 
     # 同一任务同一周期重复 claim → 40910（FR-BILL-03 发放幂等）。
     again = await client.post("/api/v1/quota/tasks/sign_in/claim", headers=headers)
@@ -108,43 +108,67 @@ async def test_claim_grants_immediately_and_repeats_are_rejected(client: Any) ->
     assert rows[0].status == "committed"
     assert rows[0].reason == "sign_in"
     assert rows[0].ref_type == "task_bonus"
-    assert rows[0].balance_after == 10 + 1
+    assert rows[0].balance_after == 10 + 10
     assert rows[0].idempotency_key.startswith(f"grant:{user_id}:sign_in:")
 
     # 幂等拒绝不改变账面。
     view = await _quota_view(client, headers, "agent_chat")
-    assert view["bonus_granted"] == 1
+    assert view["bonus_granted"] == 10
 
 
-async def test_all_five_frbill03_tasks_land_on_their_quotas(client: Any) -> None:
+async def test_final_catalog_tasks_land_on_their_quotas(client: Any) -> None:
     headers, user_id = await _login(client)
 
+    # T2b 起 report_feedback 需当日 rp_feedback 凭据、first_report 需已完成
+    # rp_report 凭据（否则 40311）：先落齐两行。本用例聚焦「各任务落到各自
+    # 配额」的发放语义，不覆盖凭据闸门（见 test_task_bonus_eligibility.py）。
+    from packages.domain.ids import next_id
+    from packages.shared.db import session_scope
+    from packages.shared.models import RpFeedback, RpReport
+
+    async with session_scope() as session:
+        report_id = next_id()
+        session.add(RpReport(report_id=report_id, user_id=user_id, status="succeeded"))
+        session.add(
+            RpFeedback(
+                feedback_id=next_id(),
+                report_id=report_id,
+                user_id=user_id,
+                dimension_key="overall",
+                verdict="accurate",
+            )
+        )
+
+    # §3.2 定稿目录（T2b）：invite_friend/profile_complete 已下架，不在 env。
     all_tasks = (
         "sign_in",
-        "profile_complete",
         "report_feedback",
         "share_report",
-        "invite_friend",
+        "first_report",
     )
     for task_key in all_tasks:
         resp = await client.post(f"/api/v1/quota/tasks/{task_key}/claim", headers=headers)
         assert resp.status_code == 200, task_key
         assert resp.json()["code"] == 0, task_key
 
-    # agent_chat 吃两个任务（sign_in + profile_complete）；tryon 吃两个；report_regen 一个。
+    # 定稿数值：sign_in→agent_chat+10、report_feedback→hairstyle+2、
+    # share_report→tryon+3、first_report→report_regen+1，各自落各自配额。
     chat = await _quota_view(client, headers, "agent_chat")
-    assert chat["bonus_granted"] == 2
-    assert chat["remaining"] == 12
+    assert chat["bonus_granted"] == 10
+    assert chat["remaining"] == 20
+    hairstyle = await _quota_view(client, headers, "hairstyle")
+    assert hairstyle["bonus_granted"] == 2
+    assert hairstyle["remaining"] == 5  # base 3 + bonus 2（task_bonus_max=2 触顶即止）
     tryon = await _quota_view(client, headers, "tryon")
-    assert tryon["bonus_granted"] == 2
-    assert tryon["remaining"] == 5  # base 3 + bonus 2（task_bonus_max=3 未触顶）
+    assert tryon["bonus_granted"] == 3
+    assert tryon["remaining"] == 6  # base 3 + bonus 3（task_bonus_max=3 未触顶）
     regen = await _quota_view(client, headers, "report_regen")
     assert regen["bonus_granted"] == 1
     assert regen["remaining"] == 2
 
     rows = await _system_grant_rows(user_id)
-    assert len(rows) == 5
-    assert len({row.idempotency_key for row in rows}) == 5
+    assert len(rows) == 4
+    assert len({row.idempotency_key for row in rows}) == 4
 
 
 async def test_unknown_task_returns_40410(client: Any) -> None:

tests/integration/test_task_bonus_eligibility.py

@@ -0,0 +1,213 @@
+"""任务换额度完成校验端到端（T2b / FR-BILL-03）：40311 凭据闸门 + 定稿数值.
+
+本机 .env 的 TASK_BONUS_RULES_JSON 可能是旧占位目录，本文件用 monkeypatch
+把规则钉死为产品方案 §3.2 定稿目录（4 项，invite_friend 下架），断言与
+.env 取值解耦。既有 test_task_bonus.py 仍按 .env 现状断言占位行为，两个
+文件各管各的口径。
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import uuid
+from collections.abc import AsyncIterator, Iterator
+from typing import Any
+
+import pytest
+
+
+def _infra_available() -> bool:
+    return bool(os.getenv("MEIDA_RUN_INTEGRATION"))
+
+
+pytestmark = [
+    pytest.mark.integration,
+    pytest.mark.skipif(
+        not _infra_available(),
+        reason="set MEIDA_RUN_INTEGRATION=1 with PostgreSQL and Redis running",
+    ),
+]
+
+#: 产品方案 §3.2 定稿目录（2026-09-11）：签到+10对话/反馈+2发型/分享+3换装/
+#: 首测+1重生成；invite_friend 不下发即下架。
+_FINAL_RULES = {
+    "sign_in": {"quota_key": "agent_chat", "amount": 10, "claim_period": "day"},
+    "report_feedback": {"quota_key": "hairstyle", "amount": 2, "claim_period": "day"},
+    "share_report": {"quota_key": "tryon", "amount": 3, "claim_period": "day"},
+    "first_report": {"quota_key": "report_regen", "amount": 1, "claim_period": "lifetime"},
+}
+
+
+@pytest.fixture
+def final_rules(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
+    """把 TASK_BONUS_RULES_JSON 钉为定稿目录（进程 env 优先于 .env）."""
+    from packages.shared.config import get_settings
+
+    monkeypatch.setenv("TASK_BONUS_RULES_JSON", json.dumps(_FINAL_RULES))
+    get_settings.cache_clear()
+    try:
+        yield
+    finally:
+        monkeypatch.undo()
+        get_settings.cache_clear()
+
+
+@pytest.fixture
+async def client() -> AsyncIterator[Any]:
+    import httpx
+
+    from apps.api.main import create_app
+    from packages.shared.config import get_settings
+
+    app = create_app(get_settings())
+    transport = httpx.ASGITransport(app=app)
+    async with (
+        httpx.AsyncClient(transport=transport, base_url="http://test") as http,
+        app.router.lifespan_context(app),
+    ):
+        yield http
+
+
+async def _login(client: Any) -> tuple[dict[str, str], int]:
+    """dev-token：每个用例用独立 device_key，拿到的就是全新用户."""
+    resp = await client.post(
+        "/api/v1/auth/dev-token", json={"device_key": f"elig-it-{uuid.uuid4().hex[:8]}"}
+    )
+    assert resp.json()["code"] == 0
+    data = resp.json()["data"]
+    return {"Authorization": f"Bearer {data['access_token']}"}, int(data["user_id"])
+
+
+async def _grant_rows(user_id: int) -> list[Any]:
+    """纯 DB 断言用：直接读 bl_ledger 的 system_grant 流水，不经过 HTTP 视图."""
+    from sqlalchemy import select
+
+    from packages.shared.db import session_scope
+    from packages.shared.models import Ledger
+
+    async with session_scope() as session:
+        result = await session.execute(
+            select(Ledger).where(Ledger.user_id == user_id, Ledger.tx_type == "system_grant")
+        )
+        return list(result.scalars().all())
+
+
+async def _create_report(user_id: int, *, status: str = "succeeded") -> int:
+    """直接落一行 rp_report 作为完成凭据（不跑流水线，只借表结构）."""
+    from packages.domain.ids import next_id
+    from packages.shared.db import session_scope
+    from packages.shared.models import RpReport
+
+    report_id = next_id()
+    async with session_scope() as session:
+        session.add(RpReport(report_id=report_id, user_id=user_id, status=status))
+    return report_id
+
+
+async def _create_feedback(user_id: int, report_id: int) -> None:
+    """落一条当日 rp_feedback（created_at 走 server_default，即“今天”）."""
+    from packages.domain.ids import next_id
+    from packages.shared.db import session_scope
+    from packages.shared.models import RpFeedback
+
+    async with session_scope() as session:
+        session.add(
+            RpFeedback(
+                feedback_id=next_id(),
+                report_id=report_id,
+                user_id=user_id,
+                dimension_key="overall",
+                verdict="accurate",
+            )
+        )
+
+
+async def test_first_report_requires_completed_report(final_rules: None, client: Any) -> None:
+    headers, user_id = await _login(client)
+
+    # 无任何报告 → 40311，且不写流水（验收：拒绝路径不落 bl_ledger）。
+    resp = await client.post("/api/v1/quota/tasks/first_report/claim", headers=headers)
+    assert resp.status_code == 403
+    assert resp.json()["code"] == 40311
+    assert await _grant_rows(user_id) == []
+
+    # 未完成（pending）的报告不算凭据。
+    await _create_report(user_id, status="pending")
+    resp = await client.post("/api/v1/quota/tasks/first_report/claim", headers=headers)
+    assert resp.status_code == 403
+    assert resp.json()["code"] == 40311
+    assert await _grant_rows(user_id) == []
+
+    # 有已完成报告 → 200，定稿数值：report_regen +1。
+    await _create_report(user_id, status="succeeded")
+    ok_resp = await client.post("/api/v1/quota/tasks/first_report/claim", headers=headers)
+    assert ok_resp.status_code == 200
+    data = ok_resp.json()["data"]
+    assert data["task_key"] == "first_report"
+    assert data["quota_key"] == "report_regen"
+    assert data["amount"] == 1
+    assert data["bonus_granted"] == 1
+
+    # lifetime 周期再 claim → 40910。
+    again = await client.post("/api/v1/quota/tasks/first_report/claim", headers=headers)
+    assert again.status_code == 409
+    assert again.json()["code"] == 40910
+    assert len(await _grant_rows(user_id)) == 1
+
+
+async def test_report_feedback_requires_same_day_feedback(final_rules: None, client: Any) -> None:
+    headers, user_id = await _login(client)
+
+    # 无反馈 → 40311，不写流水。
+    resp = await client.post("/api/v1/quota/tasks/report_feedback/claim", headers=headers)
+    assert resp.status_code == 403
+    assert resp.json()["code"] == 40311
+    assert await _grant_rows(user_id) == []
+
+    # 提交一条当日反馈后 → 200，定稿数值：hairstyle +2。
+    report_id = await _create_report(user_id)
+    await _create_feedback(user_id, report_id)
+    ok_resp = await client.post("/api/v1/quota/tasks/report_feedback/claim", headers=headers)
+    assert ok_resp.status_code == 200
+    data = ok_resp.json()["data"]
+    assert data["quota_key"] == "hairstyle"
+    assert data["amount"] == 2
+    assert data["bonus_granted"] == 2
+    assert len(await _grant_rows(user_id)) == 1
+
+
+async def test_sign_in_final_amount_and_same_day_repeat(final_rules: None, client: Any) -> None:
+    headers, _user_id = await _login(client)
+
+    resp = await client.post("/api/v1/quota/tasks/sign_in/claim", headers=headers)
+    assert resp.status_code == 200
+    data = resp.json()["data"]
+    assert data["quota_key"] == "agent_chat"
+    assert data["amount"] == 10
+    assert data["bonus_granted"] == 10
+    assert data["remaining"] == 20  # base_limit 10 + bonus 10
+
+    again = await client.post("/api/v1/quota/tasks/sign_in/claim", headers=headers)
+    assert again.status_code == 409
+    assert again.json()["code"] == 40910
+
+
+async def test_share_report_self_reported(final_rules: None, client: Any) -> None:
+    """share_report 内测期端侧自报、无服务端校验（§3.2 决策理由 4）."""
+    headers, _ = await _login(client)
+    resp = await client.post("/api/v1/quota/tasks/share_report/claim", headers=headers)
+    assert resp.status_code == 200
+    data = resp.json()["data"]
+    assert data["quota_key"] == "tryon"
+    assert data["amount"] == 3
+    assert data["bonus_granted"] == 3
+
+
+async def test_invite_friend_off_shelf_returns_40410(final_rules: None, client: Any) -> None:
+    """invite_friend 不在定稿目录：配置化下架，claim 走 40410，不写死分支."""
+    headers, user_id = await _login(client)
+    resp = await client.post("/api/v1/quota/tasks/invite_friend/claim", headers=headers)
+    assert resp.status_code == 404
+    assert resp.json()["code"] == 40410
+    assert await _grant_rows(user_id) == []

tests/integration/test_upload_direct_flow.py

@@ -0,0 +1,168 @@
+"""本地直传 → 建报告全链路（无 COS 降级路线）。
+
+dev-token 登录 → POST /uploads/direct 上传原始字节 → 拿真实 object_key →
+POST /reports 建报告走通；模拟 key 被存在性校验以 404 挡住，且不消耗配额。
+不启真 celery：API 投递钩子替换为 no-op（与 test_report_flow 同口径）。
+"""
+
+from __future__ import annotations
+
+import os
+import uuid
+from collections.abc import AsyncIterator
+from typing import Any
+
+import pytest
+
+
+def _infra_available() -> bool:
+    return bool(os.getenv("MEIDA_RUN_INTEGRATION"))
+
+
+pytestmark = [
+    pytest.mark.integration,
+    pytest.mark.skipif(
+        not _infra_available(),
+        reason="set MEIDA_RUN_INTEGRATION=1 with PostgreSQL and Redis running",
+    ),
+]
+
+JPEG_BYTES = b"\xff\xd8\xff\xe0" + b"\x00" * 64
+
+
+@pytest.fixture
+def no_enqueue(monkeypatch: pytest.MonkeyPatch) -> None:
+    """不向真 Redis 投递：本用例只验证「上传 → 建报告」入口，不跑 worker。"""
+    monkeypatch.setattr("apps.api.routers.reports._enqueue_report", lambda task_id: None)
+
+
+@pytest.fixture
+async def client(no_enqueue: None) -> AsyncIterator[Any]:
+    import httpx
+
+    from apps.api.main import create_app
+    from packages.shared.config import get_settings
+
+    app = create_app(get_settings())
+    transport = httpx.ASGITransport(app=app)
+    async with (
+        httpx.AsyncClient(transport=transport, base_url="http://test") as http,
+        app.router.lifespan_context(app),
+    ):
+        yield http
+
+
+async def _login(client: Any) -> tuple[dict[str, str], int]:
+    """dev-token：每个用例独立 device_key，拿到的就是全新用户。"""
+    resp = await client.post(
+        "/api/v1/auth/dev-token", json={"device_key": f"upload-it-{uuid.uuid4().hex[:8]}"}
+    )
+    assert resp.json()["code"] == 0
+    data = resp.json()["data"]
+    return {"Authorization": f"Bearer {data['access_token']}"}, int(data["user_id"])
+
+
+async def _upload(client: Any, headers: dict[str, str]) -> str:
+    resp = await client.post(
+        "/api/v1/uploads/direct?scene=report_photo",
+        headers={**headers, "Content-Type": "image/jpeg"},
+        content=JPEG_BYTES,
+    )
+    assert resp.status_code == 200
+    return str(resp.json()["data"]["object_key"])
+
+
+async def test_direct_upload_then_create_report_with_real_key(client: Any) -> None:
+    headers, user_id = await _login(client)
+
+    object_key = await _upload(client, headers)
+    assert object_key.startswith(f"report_photo/{user_id}/")
+    assert object_key.endswith(".jpg")
+
+    resp = await client.post(
+        "/api/v1/reports",
+        headers=headers,
+        json={
+            "questionnaire": {"style": "通勤"},
+            "photo_object_key": object_key,
+            "request_key": f"rk-{uuid.uuid4().hex}",
+        },
+    )
+    assert resp.status_code == 200
+    body = resp.json()["data"]
+    assert body["status"] == "pending"
+
+    # 落盘的输入行应记录真实 object_key。
+    from sqlalchemy import select
+
+    from packages.shared.db import session_scope
+    from packages.shared.models import RpInput
+
+    async with session_scope() as session:
+        result = await session.execute(
+            select(RpInput).where(RpInput.report_id == int(body["report_id"]))
+        )
+        rp_input = result.scalar_one()
+    assert rp_input.photo_meta == {"object_key": object_key}
+
+
+async def test_create_report_with_mock_key_is_rejected_404_without_reserving_quota(
+    client: Any,
+) -> None:
+    headers, user_id = await _login(client)
+
+    resp = await client.post(
+        "/api/v1/reports",
+        headers=headers,
+        json={
+            "questionnaire": {},
+            "photo_object_key": f"report_photo/{user_id}/mock123.jpg",  # 布局合法的模拟键
+            "request_key": f"rk-{uuid.uuid4().hex}",
+        },
+    )
+    assert resp.status_code == 404
+    assert resp.json()["code"] == 40400
+
+    # 拒绝发生在配额预扣之前：终身配额流水必须为空。
+    from sqlalchemy import func, select
+
+    from packages.shared.db import session_scope
+    from packages.shared.models import Ledger
+
+    async with session_scope() as session:
+        result = await session.execute(
+            select(func.count())
+            .select_from(Ledger)
+            .where(Ledger.user_id == user_id, Ledger.quota_key == "report_basic")
+        )
+        assert int(result.scalar_one()) == 0
+
+
+async def test_direct_upload_requires_auth(client: Any) -> None:
+    resp = await client.post(
+        "/api/v1/uploads/direct?scene=report_photo",
+        headers={"Content-Type": "image/jpeg"},
+        content=JPEG_BYTES,
+    )
+    assert resp.status_code == 401
+    assert resp.json()["code"] == 40100
+
+
+async def test_direct_upload_rejects_disallowed_type_and_scene(client: Any) -> None:
+    headers, _ = await _login(client)
+
+    bad_type = await client.post(
+        "/api/v1/uploads/direct?scene=report_photo",
+        headers={**headers, "Content-Type": "application/pdf"},
+        content=JPEG_BYTES,
+    )
+    assert bad_type.status_code == 400
+    assert bad_type.json()["code"] == 40010
+
+    bad_scene = await client.post(
+        "/api/v1/uploads/direct?scene=evil",
+        headers={**headers, "Content-Type": "image/jpeg"},
+        content=JPEG_BYTES,
+    )
+    assert bad_scene.status_code == 400
+    assert bad_scene.json()["code"] == 40010

tests/unit/test_task_bonus_rules.py

@@ -14,13 +14,13 @@ import pytest
 from packages.domain.billing.quota import PERIOD_TYPES
 from packages.shared.config import Settings
 
-#: FR-BILL-03 的五类任务：签到、完善资料、反馈、分享、邀请。
-_FR_BILL_03_TASKS = (
+#: 产品方案 §3.2 定稿目录（2026-09-11）：签到、反馈、分享、首次速测；
+#: invite_friend/profile_complete 无可信完成信号，已下架（不在默认值里）。
+_FINAL_CATALOG_TASKS = (
     "sign_in",
-    "profile_complete",
     "report_feedback",
     "share_report",
-    "invite_friend",
+    "first_report",
 )
 
 
@@ -29,10 +29,13 @@ def _rules_without_env() -> dict[str, dict[str, object]]:
     return dict(Settings(_env_file=None).task_bonus_rules)
 
 
-def test_default_rules_cover_all_five_frbill03_tasks() -> None:
+def test_default_rules_cover_the_final_catalog() -> None:
     rules = _rules_without_env()
-    for task_key in _FR_BILL_03_TASKS:
+    for task_key in _FINAL_CATALOG_TASKS:
         assert task_key in rules, task_key
+    # 下架任务不得出现在默认值里（不下发即下架，claim 返回 40410）。
+    assert "invite_friend" not in rules
+    assert "profile_complete" not in rules
 
 
 def test_every_default_rule_entry_has_a_valid_shape() -> None:
@@ -42,7 +45,7 @@ def test_every_default_rule_entry_has_a_valid_shape() -> None:
         assert isinstance(quota_key, str) and quota_key, task_key
         amount = conf.get("amount")
         assert isinstance(amount, int) and not isinstance(amount, bool), task_key
-        assert amount > 0, task_key  # 占位值 1，正式数值由产品定稿后改 env
+        assert amount > 0, task_key  # 定稿数值（§3.2），改 env 即生效
         assert conf.get("claim_period") in PERIOD_TYPES, task_key
 
 

tests/unit/test_upload_direct.py

@@ -0,0 +1,181 @@
+"""本地磁盘直传（无 COS 降级路线）：三重校验 + 落盘布局 + 存在性校验。
+
+零网络零 SDK：落盘用 tmp_path 隔离。require_local_upload 是报告链路挡
+模拟/越界 photo_object_key 的关卡，穿越、他人目录、不存在三类逐条锁死。
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from packages.shared.config import Settings
+from packages.shared.errors import AppError, ErrorCode
+from packages.storage.local import require_local_upload, save_direct_upload
+
+JPEG_BYTES = b"\xff\xd8\xff\xe0" + b"\x00" * 32
+
+
+def make_settings(tmp_path: Path, **overrides: Any) -> Settings:
+    """显式 kwargs 构造（与 test_upload_presign 同口径），测试不受 .env 影响。
+
+    upload_scenes 定义了 alias 且未开启 populate_by_name，必须按
+    ``UPLOAD_SCENES_JSON`` 传入——按字段名传会被静默忽略。
+    """
+    options: dict[str, Any] = {
+        "_env_file": None,
+        "upload_dir": str(tmp_path / "uploads"),
+        "upload_max_bytes": 1024,
+        "upload_allowed_types": "image/jpeg,image/png,image/webp",
+        "UPLOAD_SCENES_JSON": {"report_photo": {"bucket": "temp"}, "avatar": {"bucket": "temp"}},
+    }
+    options.update(overrides)
+    return Settings(**options)
+
+
+# --- save_direct_upload -----------------------------------------------------
+
+
+def test_save_writes_file_and_returns_auditable_key(tmp_path: Path) -> None:
+    settings = make_settings(tmp_path)
+    result = save_direct_upload(
+        settings, user_id=7, scene="report_photo", content_type="image/jpeg", data=JPEG_BYTES
+    )
+
+    # key 布局：{scene}/{user_id}/{uuid4hex}.{ext}（与 presign 同构，无日期段——
+    # 本地落盘没有 COS 生命周期规则需要按日分组）
+    prefix = "report_photo/7/"
+    assert result.object_key.startswith(prefix)
+    stem, ext = result.object_key.removeprefix(prefix).split(".")
+    assert ext == "jpg"
+    assert len(stem) == 32
+    assert all(char in "0123456789abcdef" for char in stem)
+
+    written = tmp_path / "uploads" / result.object_key
+    assert written.read_bytes() == JPEG_BYTES
+    assert result.size_bytes == len(JPEG_BYTES)
+    assert result.content_type == "image/jpeg"
+
+
+def test_save_maps_content_type_to_extension(tmp_path: Path) -> None:
+    settings = make_settings(tmp_path)
+    result = save_direct_upload(
+        settings, user_id=1, scene="avatar", content_type="image/webp", data=JPEG_BYTES
+    )
+    assert result.object_key.endswith(".webp")
+
+
+def test_save_rejects_unknown_scene(tmp_path: Path) -> None:
+    settings = make_settings(tmp_path)
+    with pytest.raises(AppError) as excinfo:
+        save_direct_upload(
+            settings, user_id=1, scene="evil", content_type="image/jpeg", data=JPEG_BYTES
+        )
+    assert excinfo.value.code is ErrorCode.UPLOAD_REJECTED
+    assert excinfo.value.message == "不支持的上传场景"
+
+
+def test_save_rejects_disallowed_content_type(tmp_path: Path) -> None:
+    settings = make_settings(tmp_path)
+    with pytest.raises(AppError) as excinfo:
+        save_direct_upload(
+            settings,
+            user_id=1,
+            scene="report_photo",
+            content_type="application/pdf",
+            data=JPEG_BYTES,
+        )
+    assert excinfo.value.code is ErrorCode.UPLOAD_REJECTED
+
+
+def test_save_rejects_oversize_and_empty(tmp_path: Path) -> None:
+    settings = make_settings(tmp_path)  # upload_max_bytes=1024
+    with pytest.raises(AppError) as excinfo:
+        save_direct_upload(
+            settings,
+            user_id=1,
+            scene="report_photo",
+            content_type="image/jpeg",
+            data=b"\x00" * 1025,
+        )
+    assert excinfo.value.code is ErrorCode.UPLOAD_REJECTED
+    with pytest.raises(AppError) as excinfo:
+        save_direct_upload(
+            settings, user_id=1, scene="report_photo", content_type="image/jpeg", data=b""
+        )
+    assert excinfo.value.code is ErrorCode.UPLOAD_REJECTED
+    # 拒绝路径不落任何文件
+    assert not (tmp_path / "uploads").exists()
+
+
+def test_save_size_equal_to_limit_is_allowed(tmp_path: Path) -> None:
+    settings = make_settings(tmp_path)
+    result = save_direct_upload(
+        settings,
+        user_id=1,
+        scene="report_photo",
+        content_type="image/png",
+        data=b"\x00" * 1024,
+    )
+    assert result.size_bytes == 1024
+
+
+# --- require_local_upload ---------------------------------------------------
+
+
+def test_require_roundtrip_returns_existing_file(tmp_path: Path) -> None:
+    settings = make_settings(tmp_path)
+    saved = save_direct_upload(
+        settings, user_id=7, scene="report_photo", content_type="image/jpeg", data=JPEG_BYTES
+    )
+    path = require_local_upload(
+        settings, user_id=7, scene="report_photo", object_key=saved.object_key
+    )
+    assert path.read_bytes() == JPEG_BYTES
+
+
+def test_require_rejects_missing_file(tmp_path: Path) -> None:
+    settings = make_settings(tmp_path)
+    with pytest.raises(AppError) as excinfo:
+        require_local_upload(
+            settings, user_id=7, scene="report_photo", object_key="report_photo/7/ghost.jpg"
+        )
+    assert excinfo.value.code is ErrorCode.NOT_FOUND
+
+
+def test_require_rejects_mock_key_without_upload(tmp_path: Path) -> None:
+    # 端侧历史模拟键 report_photo/{userId}/{随机}.jpg：布局合法但无落盘文件。
+    settings = make_settings(tmp_path)
+    with pytest.raises(AppError) as excinfo:
+        require_local_upload(
+            settings, user_id=7, scene="report_photo", object_key="report_photo/7/abc123.jpg"
+        )
+    assert excinfo.value.code is ErrorCode.NOT_FOUND
+
+
+def test_require_rejects_other_users_key(tmp_path: Path) -> None:
+    settings = make_settings(tmp_path)
+    saved = save_direct_upload(
+        settings, user_id=8, scene="report_photo", content_type="image/jpeg", data=JPEG_BYTES
+    )
+    with pytest.raises(AppError) as excinfo:
+        require_local_upload(settings, user_id=7, scene="report_photo", object_key=saved.object_key)
+    assert excinfo.value.code is ErrorCode.NOT_FOUND
+
+
+@pytest.mark.parametrize(
+    "evil_key",
+    [
+        "report_photo/7/../../avatar/8/x.jpg",  # 路径穿越到他人场景
+        "report_photo/7/../../../../etc/passwd",  # 穿越出 upload_dir
+        "/etc/passwd",  # 绝对路径吞噬 base
+        "avatar/7/x.jpg",  # 场景不匹配
+    ],
+)
+def test_require_rejects_out_of_bounds_keys(tmp_path: Path, evil_key: str) -> None:
+    settings = make_settings(tmp_path)
+    with pytest.raises(AppError) as excinfo:
+        require_local_upload(settings, user_id=7, scene="report_photo", object_key=evil_key)
+    assert excinfo.value.code is ErrorCode.NOT_FOUND