feat(billing): task bonus claim endpoint with idempotency and cap

POST /quota/tasks/{task_key}/claim：配置驱动任务注册表、
grant_bonus 纯域函数、ledger 幂等键防重、task_bonus_max 封顶（FR-BILL-03/04）。

实现：packages/domain/billing/quota.py 新增 BonusDecision/grant_bonus（mypy strict，
恰好到顶允许、超一格整体拒绝）、grant_idempotency_key、PERIOD_TYPES；
packages/billing/service.py 新增 claim_task_bonus（幂等快捷路径 → 行锁 → 封顶判定
→ 同事务更新 bonus_granted 并写 system_grant 流水，delta=0 且 balance_after 按
remaining 口径；IntegrityError 并发兜底回滚为 40910）；响应模型 TaskBonusClaimResponse
放新建的 packages/billing/schemas.py；路由挂在 apps/api/routers/quota.py，
rules 由 settings.task_bonus_rules 注入（amount 为占位值，产品定稿后改 env）。
测试：新增单测 11 个（封顶边界、amount<=0、不动 used、hard_cap 独立生效、
默认规则形状与 env 覆盖通路）；新增集成 4 个（领取+重复 40910+纯 DB 流水断言、
五类任务全部到账、未知任务 40410、amount=11>max=10 → 40310 且无新流水）。
门禁全绿：ruff check / ruff format --check / mypy 两段 / lint-imports；
单测 180 passed，集成 7 passed。

偏差说明：1) 任务文档提到的 BillingService._load_rule 实际名为 _require_rule，
   直接复用现有助手；2) packages/billing/repo.py 无需新增函数（现有
   get_ledger_by_idempotency_key/lock_or_create_quota/add_ledger 已覆盖本任务）；
   3) 文档设想的「TASK_BONUS_RULES_JSON 空值容错」经 env 源实测不成立——
   pydantic-settings 对 dict 字段先做 JSON 解码，空白字符串在 Settings 构造时即
   报错（共享校验器/config.py 属禁改，未动）；单测改为覆盖受支持的空配置形态 "{}"。

apps/api/routers/quota.py

@@ -1,4 +1,4 @@
-"""``GET /quota`` and ``GET /quota/ledger`` (架构文档 §9.3 配额).
+"""``GET /quota``、``GET /quota/ledger``、``POST /quota/tasks/{task_key}/claim``.
 
 FR-BILL-07: 额度相关文案中**不得出现任何价格、购买、会员、升级字样**。
 Only two recovery paths may ever be surfaced: 等待恢复 and 做任务换额度.
@@ -6,11 +6,12 @@ Only two recovery paths may ever be surfaced: 等待恢复 and 做任务换额
 
 from __future__ import annotations
 
-from fastapi import APIRouter, Query
+from fastapi import APIRouter, Path, Query
 
-from apps.api.deps import BillingServiceDep, CurrentUserId
+from apps.api.deps import BillingServiceDep, CurrentUserId, SettingsDep
 from apps.api.schemas import LedgerEntry, QuotaResponse
 from packages.agent.service import QUOTA_KEY_AGENT_CHAT
+from packages.billing.schemas import TaskBonusClaimResponse
 from packages.shared.response import ok
 
 router = APIRouter(prefix="/quota", tags=["quota"])
@@ -59,3 +60,31 @@ async def get_ledger(
             for row in rows
         ]
     )
+
+
+@router.post("/tasks/{task_key}/claim", summary="做任务换额度：领取任务奖励（FR-BILL-03）")
+async def claim_task_bonus(
+    user_id: CurrentUserId,
+    billing: BillingServiceDep,
+    settings: SettingsDep,
+    task_key: str = Path(max_length=64),
+) -> dict[str, object]:
+    """同一任务同一周期只能领一次（重复领返回 40910），受规则封顶（40310）.
+
+    奖励数值由 ``TASK_BONUS_RULES_JSON`` 配置驱动（FR-BILL-04），当前是占位值，
+    产品定稿后改 env 即生效。
+    """
+    claim = await billing.claim_task_bonus(
+        user_id=user_id,
+        task_key=task_key,
+        rules=dict(settings.task_bonus_rules),
+    )
+    return ok(
+        TaskBonusClaimResponse(
+            task_key=claim.task_key,
+            quota_key=claim.quota_key,
+            amount=claim.amount,
+            bonus_granted=claim.bonus_granted,
+            remaining=claim.remaining,
+        ).model_dump()
+    )

packages/billing/schemas.py

@@ -0,0 +1,20 @@
+"""billing 域对外响应模型（FR-BILL-03 任务奖励）.
+
+``apps/api/schemas.py`` 是全局共享契约文件，本轮并行开发禁改；billing 新增
+接口的响应模型放本包，由路由层装配进 ``ok(...)`` 信封。字段口径与
+NFR-MAINT-01「契约先行」一致：客户端类型从 OpenAPI 生成。
+"""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class TaskBonusClaimResponse(BaseModel):
+    """``POST /quota/tasks/{task_key}/claim`` 的 data 载荷（FR-BILL-03）."""
+
+    task_key: str = Field(max_length=64, description="任务标识，如 sign_in")
+    quota_key: str = Field(max_length=64, description="奖励落入的额度种类")
+    amount: int = Field(description="本次发放数额（当前为占位值，正式数值由产品定稿后改配置）")
+    bonus_granted: int = Field(description="本周期累计已获得的任务奖励")
+    remaining: int = Field(description="发放后该额度的剩余次数")

packages/billing/service.py

@@ -11,8 +11,9 @@ from __future__ import annotations
 
 from dataclasses import dataclass
 from datetime import UTC, datetime
-from typing import cast
+from typing import Any, cast
 
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from packages.billing import repo
@@ -49,6 +50,43 @@ class QuotaSnapshot:
     reset_at: datetime | None
 
 
+@dataclass(frozen=True, slots=True)
+class TaskBonusClaim:
+    """一次成功领取的结果（路由层据此组装 TaskBonusClaimResponse）."""
+
+    task_key: str
+    quota_key: str
+    amount: int
+    bonus_granted: int
+    remaining: int
+
+
+def _parse_task_rule(
+    task_key: str, conf: dict[str, Any]
+) -> tuple[str, int, quota_rules.PeriodType]:
+    """解析 ``settings.task_bonus_rules`` 的一条配置.
+
+    坏配置是运维错误（5xxxx），不是用户错误：提前在这里拦下，比让
+    ``period_key`` 在请求路径上抛 ValueError 变成 500 清晰得多。
+    """
+    quota_key = conf.get("quota_key")
+    amount = conf.get("amount")
+    claim_period = conf.get("claim_period")
+    if not isinstance(quota_key, str) or not quota_key:
+        raise AppError(
+            ErrorCode.CONFIG_ERROR, message="任务奖励配置不合法", data={"task_key": task_key}
+        )
+    if not isinstance(amount, int) or isinstance(amount, bool) or amount <= 0:
+        raise AppError(
+            ErrorCode.CONFIG_ERROR, message="任务奖励配置不合法", data={"task_key": task_key}
+        )
+    if not isinstance(claim_period, str) or claim_period not in quota_rules.PERIOD_TYPES:
+        raise AppError(
+            ErrorCode.CONFIG_ERROR, message="任务奖励配置不合法", data={"task_key": task_key}
+        )
+    return quota_key, amount, cast(quota_rules.PeriodType, claim_period)
+
+
 class BillingService:
     def __init__(self, session: AsyncSession, *, tz_offset_hours: int = 8) -> None:
         self._session = session
@@ -263,6 +301,119 @@ class BillingService:
             remaining=new_state.remaining,
         )
 
+    # -- task bonus (FR-BILL-03) --------------------------------------------
+
+    async def claim_task_bonus(
+        self,
+        *,
+        user_id: int,
+        task_key: str,
+        rules: dict[str, dict[str, Any]],
+    ) -> TaskBonusClaim:
+        """领取任务奖励：发放即到账、幂等、封顶（FR-BILL-03）.
+
+        ``rules`` 来自 ``settings.task_bonus_rules``（FR-BILL-04 配置化）：
+        amount 目前是占位值，产品在产品方案 §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)
+
+        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)
+        claim_pk = quota_rules.period_key(now, claim_period, self._tz_offset)
+        idem_key = quota_rules.grant_idempotency_key(
+            user_id=user_id, task_key=task_key, claim_period_key=claim_pk
+        )
+
+        # 幂等快捷路径：本周期已发放过就直接拒绝，不进任何写路径。
+        if await repo.get_ledger_by_idempotency_key(self._session, idem_key) is not None:
+            QUOTA_EVENTS.labels(quota_key, "grant", "idempotent_hit").inc()
+            raise AppError(ErrorCode.TASK_BONUS_ALREADY_CLAIMED, data={"task_key": task_key})
+
+        row = await repo.lock_or_create_quota(
+            self._session,
+            user_id=user_id,
+            quota_key=quota_key,
+            period_key=quota_pk,
+            limit_value=rule.base_limit,
+        )
+        state = quota_rules.QuotaState(
+            used=row.used,
+            base_limit=row.limit_value,
+            bonus=row.bonus_granted,
+            hard_cap=rule.hard_cap,
+        )
+        decision = quota_rules.grant_bonus(state, amount, rule.task_bonus_max)
+        if not decision.allowed:
+            # 封顶拒绝：不写任何流水，bonus_granted 保持原值（验收 3）。
+            QUOTA_EVENTS.labels(quota_key, "grant", "capped").inc()
+            raise AppError(
+                ErrorCode.TASK_BONUS_CAP_REACHED,
+                data={
+                    "task_key": task_key,
+                    "bonus_granted": state.bonus,
+                    "task_bonus_max": rule.task_bonus_max,
+                },
+            )
+
+        row.bonus_granted = decision.bonus
+        tx_id = next_id()
+        try:
+            repo.add_ledger(
+                self._session,
+                Ledger(
+                    tx_id=tx_id,
+                    user_id=user_id,
+                    # delta=0：system_grant 不改变 used，只提高有效额度；balance_after
+                    # 仍按 remaining 语义记录 —— 流水可重算的是 used，不是余额本身，
+                    # 与 reserve/commit 的记账口径一致。
+                    delta=0,
+                    tx_type="system_grant",
+                    status="committed",
+                    reason=task_key,
+                    ref_type="task_bonus",
+                    ref_id=task_key,
+                    quota_key=quota_key,
+                    period_key=quota_pk,
+                    balance_after=decision.remaining,
+                    idempotency_key=idem_key,
+                ),
+            )
+            await self._session.flush()
+        except IntegrityError as exc:
+            # 并发兜底：两个请求同时通过上面的快捷检查，idempotency_key 唯一约束
+            # 让后提交者失败。对用户来说结果与快捷路径等价 —— 已领取，而非 500。
+            await self._session.rollback()
+            QUOTA_EVENTS.labels(quota_key, "grant", "idempotent_hit").inc()
+            raise AppError(
+                ErrorCode.TASK_BONUS_ALREADY_CLAIMED, data={"task_key": task_key}
+            ) from exc
+
+        QUOTA_EVENTS.labels(quota_key, "grant", "ok").inc()
+        logger.info(
+            "task_bonus_granted",
+            tx_id=tx_id,
+            task_key=task_key,
+            quota_key=quota_key,
+            amount=amount,
+            bonus=decision.bonus,
+            remaining=decision.remaining,
+        )
+        return TaskBonusClaim(
+            task_key=task_key,
+            quota_key=quota_key,
+            amount=amount,
+            bonus_granted=decision.bonus,
+            remaining=decision.remaining,
+        )
+
     # -- helpers ------------------------------------------------------------
 
     async def _require_rule(self, quota_key: str) -> QuotaRule:

packages/domain/billing/quota.py

@@ -15,11 +15,13 @@ from __future__ import annotations
 
 from dataclasses import dataclass, replace
 from datetime import UTC, datetime, timedelta, timezone
-from typing import Literal
+from typing import Literal, get_args
 
 from packages.domain.errors import QuotaExceededError
 
 PeriodType = Literal["day", "month", "lifetime"]
+#: 合法周期取值集合，供配置校验复用；单一事实来源是上面的 Literal。
+PERIOD_TYPES: frozenset[str] = frozenset(get_args(PeriodType))
 LedgerStatus = Literal["reserved", "committed", "refunded"]
 TxType = Literal["system_grant", "consume", "refund", "purchase", "gift"]
 
@@ -125,6 +127,51 @@ def refund(state: QuotaState, amount: int) -> QuotaState:
     return replace(state, used=max(state.used - amount, 0))
 
 
+@dataclass(frozen=True, slots=True)
+class BonusDecision:
+    """任务奖励发放的封顶判定结果（FR-BILL-03）.
+
+    ``allowed=False`` 时 ``bonus/remaining/effective_limit`` 保持发放前的值：
+    调用方必须整体放弃，不做部分发放 ——「做任务换 N 次」对用户是明确承诺，
+    发一半既说不清也无法用流水对账。
+    """
+
+    allowed: bool
+    amount: int
+    bonus: int
+    remaining: int
+    effective_limit: int
+
+
+def grant_bonus(state: QuotaState, amount: int, task_bonus_max: int) -> BonusDecision:
+    """把任务奖励 ``amount`` 计入 ``state.bonus``，超上限则整体拒绝（FR-BILL-03）.
+
+    - 只动 ``bonus``，不动 ``used``：任务奖励抬高的是额度，不是用量；
+    - ``task_bonus_max`` 来自 ``bl_quota_rule``，是任务奖励独立于 ``hard_cap``
+      的第二道天花板（两者都生效时 effective_limit 取更紧的）；
+    - 边界语义：恰好到顶（bonus + amount == max）允许，超出一格（== max+1）拒绝。
+    """
+    if amount <= 0:
+        msg = "amount must be positive"
+        raise ValueError(msg)
+    if state.bonus + amount > task_bonus_max:
+        return BonusDecision(
+            allowed=False,
+            amount=amount,
+            bonus=state.bonus,
+            remaining=state.remaining,
+            effective_limit=state.effective_limit,
+        )
+    granted = replace(state, bonus=state.bonus + amount)
+    return BonusDecision(
+        allowed=True,
+        amount=amount,
+        bonus=granted.bonus,
+        remaining=granted.remaining,
+        effective_limit=granted.effective_limit,
+    )
+
+
 def balance_after(state: QuotaState) -> int:
     """Value written to ``bl_ledger.balance_after``.
 
@@ -139,6 +186,15 @@ def reserve_idempotency_key(*, user_id: int, quota_key: str, request_key: str) -
     return f"rsv:{user_id}:{quota_key}:{request_key}"
 
 
+def grant_idempotency_key(*, user_id: int, task_key: str, claim_period_key: str) -> str:
+    """FR-BILL-03 发放幂等：同一任务同一周期派生同一个键.
+
+    周期取任务的 ``claim_period``（不是额度行的周期）——「每天可领」的月度
+    额度任务，幂等粒度必须跟任务承诺走。
+    """
+    return f"grant:{user_id}:{task_key}:{claim_period_key}"
+
+
 def refund_idempotency_key(reserve_key: str) -> str:
     """FR-BILL-05: 退还必须幂等。Derived, so a retry cannot double-refund."""
     return f"rfd:{reserve_key}"

tests/integration/test_task_bonus.py

@@ -0,0 +1,190 @@
+"""任务奖励领取端到端（FR-BILL-03）：发放、幂等、封顶、流水.
+
+只依赖 PostgreSQL（额度与账本都在库里），不经过 Redis，也不触发任何 LLM
+调用 —— 与 test_vertical_slice.py 的 FakeProvider 思路一致：集成用例验证
+配额链路，不该花钱也不该受网络影响。
+"""
+
+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",
+    ),
+]
+
+
+@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"bonus-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 _system_grant_rows(user_id: int) -> list[Any]:
+    """纯 DB 断言用：直接读 bl_ledger，不经过 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 _quota_view(client: Any, headers: dict[str, str], quota_key: str) -> dict[str, Any]:
+    resp = await client.get("/api/v1/quota", headers=headers, params={"quota_key": quota_key})
+    assert resp.json()["code"] == 0
+    data: dict[str, Any] = resp.json()["data"]
+    return data
+
+
+async def test_claim_grants_immediately_and_repeats_are_rejected(client: Any) -> None:
+    headers, user_id = await _login(client)
+
+    claim = await client.post("/api/v1/quota/tasks/sign_in/claim", headers=headers)
+    assert claim.status_code == 200
+    body = claim.json()
+    assert body["code"] == 0
+    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
+    }
+
+    # 发放即到账：额度视图立刻可见。
+    view = await _quota_view(client, headers, "agent_chat")
+    assert view["bonus_granted"] == 1
+    assert view["remaining"] == 11
+
+    # 同一任务同一周期重复 claim → 40910（FR-BILL-03 发放幂等）。
+    again = await client.post("/api/v1/quota/tasks/sign_in/claim", headers=headers)
+    assert again.status_code == 409
+    assert again.json()["code"] == 40910
+
+    # 纯 DB 断言：恰好一条 system_grant 流水，balance_after = base + bonus。
+    rows = await _system_grant_rows(user_id)
+    assert len(rows) == 1
+    assert rows[0].delta == 0
+    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].idempotency_key.startswith(f"grant:{user_id}:sign_in:")
+
+    # 幂等拒绝不改变账面。
+    view = await _quota_view(client, headers, "agent_chat")
+    assert view["bonus_granted"] == 1
+
+
+async def test_all_five_frbill03_tasks_land_on_their_quotas(client: Any) -> None:
+    headers, user_id = await _login(client)
+
+    all_tasks = (
+        "sign_in",
+        "profile_complete",
+        "report_feedback",
+        "share_report",
+        "invite_friend",
+    )
+    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 一个。
+    chat = await _quota_view(client, headers, "agent_chat")
+    assert chat["bonus_granted"] == 2
+    assert chat["remaining"] == 12
+    tryon = await _quota_view(client, headers, "tryon")
+    assert tryon["bonus_granted"] == 2
+    assert tryon["remaining"] == 5  # base 3 + bonus 2（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
+
+
+async def test_unknown_task_returns_40410(client: Any) -> None:
+    headers, _ = await _login(client)
+    resp = await client.post("/api/v1/quota/tasks/not_a_real_task/claim", headers=headers)
+    assert resp.status_code == 404
+    assert resp.json()["code"] == 40410
+
+
+@pytest.fixture
+def capped_rules(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
+    """把 sign_in 的 amount 覆盖成 11（> agent_chat 的 task_bonus_max=10）.
+
+    monkeypatch.setenv + get_settings.cache_clear()（结束恢复），与 T1 的
+    settings 覆盖写法一致；依赖该 fixture 的用例会把 client 建在覆盖之后。
+    """
+    from packages.shared.config import get_settings
+
+    monkeypatch.setenv(
+        "TASK_BONUS_RULES_JSON",
+        json.dumps({"sign_in": {"quota_key": "agent_chat", "amount": 11, "claim_period": "day"}}),
+    )
+    get_settings.cache_clear()
+    try:
+        yield
+    finally:
+        monkeypatch.undo()
+        get_settings.cache_clear()
+
+
+async def test_claim_over_the_cap_is_rejected_without_any_ledger(
+    capped_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 == 403
+    assert resp.json()["code"] == 40310
+
+    # 拒绝路径不写任何流水，bonus 也不动（FR-BILL-03 验收 3）。
+    assert await _system_grant_rows(user_id) == []
+    view = await _quota_view(client, headers, "agent_chat")
+    assert view["bonus_granted"] == 0

tests/unit/test_bonus_domain.py

@@ -0,0 +1,74 @@
+"""grant_bonus 纯域函数：封顶边界与不变量（FR-BILL-03）."""
+
+from __future__ import annotations
+
+import pytest
+
+from packages.domain.billing import quota
+
+
+def test_grant_up_to_the_cap_exactly_is_allowed() -> None:
+    # bonus + amount == max：恰好到顶，允许。
+    decision = quota.grant_bonus(quota.QuotaState(used=0, base_limit=10, bonus=9), 1, 10)
+    assert decision.allowed
+    assert decision.bonus == 10
+    assert decision.effective_limit == 20
+    assert decision.remaining == 20
+
+
+def test_grant_one_over_the_cap_is_rejected_as_a_whole() -> None:
+    # bonus + amount == max + 1：整体拒绝，不做部分发放。
+    state = quota.QuotaState(used=4, base_limit=10, bonus=9)
+    decision = quota.grant_bonus(state, 2, 10)
+    assert not decision.allowed
+    # 拒绝时保持发放前的值 —— 调用方不得据此落库。
+    assert decision.bonus == 9
+    assert decision.remaining == state.remaining
+    assert decision.effective_limit == state.effective_limit
+
+
+def test_zero_cap_rejects_every_grant() -> None:
+    # hd_export / report_basic 的 task_bonus_max=0：任务奖励一律拒绝。
+    decision = quota.grant_bonus(quota.QuotaState(used=0, base_limit=1), 1, 0)
+    assert not decision.allowed
+
+
+def test_grant_rejects_non_positive_amounts() -> None:
+    state = quota.QuotaState(used=0, base_limit=10)
+    with pytest.raises(ValueError, match="positive"):
+        quota.grant_bonus(state, 0, 10)
+    with pytest.raises(ValueError, match="positive"):
+        quota.grant_bonus(state, -1, 10)
+
+
+def test_grant_never_touches_used_or_base_limit() -> None:
+    # 任务奖励抬高的是额度不是用量：remaining 的增量恰好等于 amount。
+    state = quota.QuotaState(used=7, base_limit=10, bonus=2)
+    decision = quota.grant_bonus(state, 3, 10)
+    assert decision.allowed
+    assert decision.remaining - state.remaining == 3
+    assert decision.bonus == 5
+
+
+def test_task_bonus_cap_is_independent_of_hard_cap() -> None:
+    # 两道天花板独立生效：task_bonus_max 放行、hard_cap 把 effective_limit 削回来。
+    state = quota.QuotaState(used=0, base_limit=3, bonus=3, hard_cap=5)
+    decision = quota.grant_bonus(state, 3, 10)
+    assert decision.allowed  # bonus 6 <= task_bonus_max 10
+    assert decision.effective_limit == 5  # base 3 + bonus 6 被 hard_cap 削到 5
+    assert decision.remaining == 5
+
+
+def test_grant_key_is_derived_per_user_task_and_claim_period() -> None:
+    key = quota.grant_idempotency_key(
+        user_id=7, task_key="sign_in", claim_period_key="d:2026-08-14"
+    )
+    assert key == "grant:7:sign_in:d:2026-08-14"
+    # 派生键确定：同一周期重复 claim 得到同一个键，唯一约束才能兜底。
+    assert key == quota.grant_idempotency_key(
+        user_id=7, task_key="sign_in", claim_period_key="d:2026-08-14"
+    )
+    # 换周期换键：下个周期可以再领。
+    assert key != quota.grant_idempotency_key(
+        user_id=7, task_key="sign_in", claim_period_key="d:2026-08-15"
+    )

tests/unit/test_task_bonus_rules.py

@@ -0,0 +1,62 @@
+"""settings.task_bonus_rules 的形状与 env 通路（FR-BILL-03/04）.
+
+协调提交已把 ``task_bonus_rules`` 纳入 ``_parse_json_object`` 共享校验器，
+解析逻辑不再单独测；这里测默认值形状（每条配置必须能被 claim 路径安全消费）
+以及集成测试要依赖的 ``TASK_BONUS_RULES_JSON`` 覆盖通路。
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from packages.domain.billing.quota import PERIOD_TYPES
+from packages.shared.config import Settings
+
+#: FR-BILL-03 的五类任务：签到、完善资料、反馈、分享、邀请。
+_FR_BILL_03_TASKS = (
+    "sign_in",
+    "profile_complete",
+    "report_feedback",
+    "share_report",
+    "invite_friend",
+)
+
+
+def _rules_without_env() -> dict[str, dict[str, object]]:
+    # _env_file=None：不读 .env，只看进程环境，保证断言的是代码内默认值。
+    return dict(Settings(_env_file=None).task_bonus_rules)
+
+
+def test_default_rules_cover_all_five_frbill03_tasks() -> None:
+    rules = _rules_without_env()
+    for task_key in _FR_BILL_03_TASKS:
+        assert task_key in rules, task_key
+
+
+def test_every_default_rule_entry_has_a_valid_shape() -> None:
+    for task_key, conf in _rules_without_env().items():
+        assert isinstance(conf, dict), task_key
+        quota_key = conf.get("quota_key")
+        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 conf.get("claim_period") in PERIOD_TYPES, task_key
+
+
+def test_rules_json_env_reaches_settings_via_alias(monkeypatch: pytest.MonkeyPatch) -> None:
+    # 集成测试用同一通路把 amount 覆盖成超限值验证封顶，这里先证明通路本身可用。
+    raw = json.dumps({"sign_in": {"quota_key": "agent_chat", "amount": 11, "claim_period": "day"}})
+    monkeypatch.setenv("TASK_BONUS_RULES_JSON", raw)
+    settings = Settings(_env_file=None)
+    assert settings.task_bonus_rules["sign_in"]["amount"] == 11
+
+
+def test_empty_rules_object_env_means_no_tasks(monkeypatch: pytest.MonkeyPatch) -> None:
+    # 关闭任务奖励的受支持形态是 "{}"：解析为空注册表，所有 claim 返回 40410，
+    # 服务照常启动。（空白字符串不是合法 JSON，pydantic-settings 在 Settings
+    # 构造时就地失败——那是共享校验器的行为，不归本任务改。）
+    monkeypatch.setenv("TASK_BONUS_RULES_JSON", "{}")
+    assert Settings(_env_file=None).task_bonus_rules == {}