diff --git a/app/api/deps.py b/app/api/deps.py
new file mode 100644
index 0000000..409f73e
--- /dev/null
+++ b/app/api/deps.py
@@ -0,0 +1,28 @@
+"""API 依赖：鉴权上下文（架构 §5.7 · 归属校验统一在此层）。
+
+AuthContext 模型在 A4 冻结（开发计划 v1.1）；get_auth_context 完整实现归 B6：
+dev（app_env=development）从 X-Debug-Role / X-Debug-Actor 请求头构造，
+非 dev 环境启动时检测 debug 依赖注册即拒绝；T-01 就绪后仅替换工厂内部为 JWT 解析。
+"""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class AuthContext(BaseModel):
+    """统一鉴权上下文（全部 API 依赖层的产出；service 层签名接收此类型）。"""
+
+    actor_id: str = Field(..., description="操作者 ID：staff_id 或 customer_id")
+    roles: list[str] = Field(default_factory=list, description="角色集合，如 ['risk_officer','risk_demo']")
+    customer_id: str | None = Field(None, description="customer 角色时 = 本人 customer_id；其余为空")
+
+    def has_role(self, *roles: str) -> bool:
+        return any(r in self.roles for r in roles)
+
+    def is_customer(self) -> bool:
+        return "customer" in self.roles
+
+
+def get_auth_context() -> AuthContext:
+    raise NotImplementedError("implemented in B6 (X-Debug-* in dev / JWT in T-01)")
diff --git a/app/service/suitability.py b/app/service/suitability.py
new file mode 100644
index 0000000..df28cb0
--- /dev/null
+++ b/app/service/suitability.py
@@ -0,0 +1,157 @@
+"""适当性校验服务（R-02 · 全系统唯一阻断点 · SUIT-001~008）。
+
+规则权威：docs/PRD/附-风控规则表.md §1。返回语义（PRD FR-2）：
+- is_matched = 纯等级矩阵结果（含 SUIT-006 封顶后判定）
+- blocked = 最终是否阻断（= NOT is_matched 或 SUIT-008 测评过期）
+- reasons[] 列明各规则判定；落库 customer_risk_level 记**原测评等级**
+- 数据归属校验（G-01）在 FastAPI 依赖层完成，本服务不做
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import date, datetime
+from decimal import Decimal
+from typing import Any
+
+from app.repository.core_ro import CoreReadOnlyRepository
+from app.repository.risk_repository import RiskRepository
+from app.utils.trace import current_trace
+
+TIER_ORDER = {"C1": 1, "C2": 2, "C3": 3, "C4": 4, "C5": 5}
+
+
+@dataclass(frozen=True)
+class SuitabilityResult:
+    customer_level: str  # 原测评等级（落库用）
+    effective_level: str  # SUIT-006 封顶后判定等级
+    product_level: str
+    is_matched: bool
+    blocked: bool
+    reasons: list[str] = field(default_factory=list)
+    block_reason: str = ""  # 阻断主因（人类可读，不含文案模板）
+
+
+def grade_number(grade: str) -> int:
+    """C1~C5 / R1~R5 → 1~5；非法值抛错（数据污染即 Fail-fast）。"""
+    n = TIER_ORDER.get(grade)
+    if n is None:
+        n = TIER_ORDER.get(grade.replace("R", "C"))
+    if n is None:
+        raise ValueError(f"invalid risk grade: {grade!r}")
+    return n
+
+
+def cap_by_age(level: str, age: int | None) -> tuple[str, str | None]:
+    """SUIT-006：年龄 ≥70 按最高 C3 封顶；age IS NULL 跳过并提示人工复核。"""
+    if age is None:
+        return level, "年龄缺失，建议人工复核"
+    if age >= 70 and grade_number(level) > grade_number("C3"):
+        return "C3", f"客户测评 {level} 因年龄≥70 按 C3 处理（SUIT-006）"
+    return level, None
+
+
+def is_assessment_valid(evaluated_at: date | datetime | str | None, today: date, valid_days: int) -> bool:
+    """SUIT-008：测评有效期默认 365 天（.env 可配）。兼容 str 输入（驱动差异防御）。"""
+    if evaluated_at is None:
+        return False
+    if isinstance(evaluated_at, str):
+        evaluated_at = date.fromisoformat(evaluated_at[:10])
+    if isinstance(evaluated_at, datetime):
+        evaluated_at = evaluated_at.date()
+    return (today - evaluated_at).days < valid_days
+
+
+def match_by_matrix(customer_level: str, product_level: str) -> bool:
+    """SUIT-001~005：客户等级序号 ≥ 产品等级序号方可购买。"""
+    return grade_number(customer_level) >= grade_number(product_level)
+
+
+def check_core(
+    customer: dict[str, Any],
+    product: dict[str, Any],
+    valid_days: int = 365,
+    today: date | None = None,
+) -> SuitabilityResult:
+    """纯函数核心（无 IO）：customer=core_customer+risk 合并行，product=core_product 行。
+
+    键约定：customer.risk_code（C1~C5）、customer.age（可 None）、
+    customer.risk_evaluated_at（date/datetime）；product.min_risk_code（R1~R5）。
+    """
+    today = today or date.today()
+    raw_level: str = customer["risk_code"]
+    product_level: str = product["min_risk_code"]
+    reasons: list[str] = []
+
+    effective_level, cap_reason = cap_by_age(raw_level, customer.get("age"))
+    if cap_reason:
+        reasons.append(cap_reason)
+
+    is_matched = match_by_matrix(effective_level, product_level)
+    if is_matched:
+        reasons.append(f"等级矩阵通过：{effective_level} 可购 {product_level}（SUIT-001~005）")
+    else:
+        reasons.append(f"等级矩阵不通过：{effective_level} 不可购 {product_level}（SUIT-001~005）")
+
+    expired = not is_assessment_valid(customer.get("risk_evaluated_at"), today, valid_days)
+    if expired:
+        reasons.append(f"风险测评已过期（有效期 {valid_days} 天），请重新测评（SUIT-008）")
+
+    blocked = (not is_matched) or expired
+    if not is_matched:
+        block_reason = f"您的风险等级为{raw_level}（判定按{effective_level}），该产品为{product_level}，风险不匹配"
+    elif expired:
+        block_reason = "您的风险测评已过期，无法购买新产品，请重新测评"
+    else:
+        block_reason = ""
+
+    return SuitabilityResult(
+        customer_level=raw_level,
+        effective_level=effective_level,
+        product_level=product_level,
+        is_matched=is_matched,
+        blocked=blocked,
+        reasons=reasons,
+        block_reason=block_reason,
+    )
+
+
+def suitability_check(
+    customer_id: str,
+    product_id: str,
+    core_ro: CoreReadOnlyRepository | None = None,
+    risk_repo: RiskRepository | None = None,
+    valid_days: int = 365,
+    today: date | None = None,
+) -> SuitabilityResult:
+    """服务入口：查 L0 事实 → 纯函数判定 → 落 risk_suitability_log（每次校验必落）。
+
+    阻断时的预警单由调用方（交易网关 FR-1）生成，本函数只落校验日志。
+    """
+    core_ro = core_ro or CoreReadOnlyRepository()
+    risk_repo = risk_repo or RiskRepository()
+
+    customer = core_ro.get_customer_l0(customer_id)
+    if customer is None:
+        raise LookupError(f"customer not found: {customer_id}")
+    product = core_ro.get_product(product_id)
+    if product is None:
+        raise LookupError(f"product not found: {product_id}")
+
+    result = check_core(customer, product, valid_days=valid_days, today=today)
+
+    risk_repo.insert_suitability_log(
+        {
+            "trace_id": current_trace(),
+            "customer_id": customer_id,
+            "product_id": product_id,
+            "customer_risk_level": result.customer_level,
+            "product_risk_level": result.product_level,
+            "is_matched": int(result.is_matched),
+            "is_blocked": int(result.blocked),
+            "block_reason": result.block_reason or None,
+            "request_ref": None,
+            "profile_l1_version": None,
+        }
+    )
+    return result
diff --git a/tests/test_suitability.py b/tests/test_suitability.py
new file mode 100644
index 0000000..fb1758a
--- /dev/null
+++ b/tests/test_suitability.py
@@ -0,0 +1,198 @@
+"""suitability 单测（A4 · 验收 A-8）：矩阵 25 组合 + SUIT-006/008 边界 + 服务集成。"""
+
+from datetime import date, datetime, timedelta
+
+import pytest
+from sqlalchemy import create_engine, text
+
+from app.repository.core_ro import CoreReadOnlyRepository
+from app.repository.risk_repository import RiskRepository
+from app.service.suitability import (
+    cap_by_age,
+    check_core,
+    grade_number,
+    is_assessment_valid,
+    match_by_matrix,
+    suitability_check,
+)
+
+TODAY = date(2026, 9, 6)
+
+
+def _customer(level, age=40, evaluated=TODAY - timedelta(days=90)):
+    return {
+        "customer_id": "C1",
+        "risk_code": level,
+        "age": age,
+        "risk_evaluated_at": evaluated,
+    }
+
+
+def _product(level):
+    return {"product_id": "P1", "min_risk_code": level}
+
+
+class TestMatrix25:
+    """SUIT-001~005：C1~C5 × R1~R5 全组合。"""
+
+    @pytest.mark.parametrize("c", ["C1", "C2", "C3", "C4", "C5"])
+    @pytest.mark.parametrize("p", ["R1", "R2", "R3", "R4", "R5"])
+    def test_matrix(self, c, p):
+        expect_match = grade_number(c) >= grade_number(p)
+        result = check_core(_customer(c), _product(p), today=TODAY)
+        assert result.is_matched is expect_match
+        assert result.blocked is (not expect_match)  # 测评有效期内 blocked ≡ not matched
+        assert result.customer_level == c and result.effective_level == c
+
+    def test_a1_c1_buys_r4(self):
+        """验收 A-1 断言口径：C1+R4 → 不匹配 + 阻断 + 原因含 SUIT-001~005。"""
+        result = check_core(_customer("C1"), _product("R4"), today=TODAY)
+        assert result.is_matched is False and result.blocked is True
+        assert "SUIT-001~005" in result.reasons[-1]
+
+
+class TestSUIT006AgeCap:
+    def test_70yo_c5_capped_to_c3(self):
+        level, reason = cap_by_age("C5", 70)
+        assert level == "C3" and "SUIT-006" in reason
+
+    def test_69yo_c5_not_capped(self):
+        assert cap_by_age("C5", 69) == ("C5", None)
+
+    def test_70yo_c3_unchanged(self):
+        assert cap_by_age("C3", 72) == ("C3", None)
+
+    def test_null_age_skipped_with_review_hint(self):
+        level, reason = cap_by_age("C5", None)
+        assert level == "C5" and reason == "年龄缺失，建议人工复核"
+
+    def test_a2_70yo_c5_buys_r4_blocked(self):
+        """验收 A-2 断言口径：70 岁 C5 买 R4 → 封顶说明 + 不匹配，无过期干扰。"""
+        result = check_core(_customer("C5", age=70), _product("R4"), today=TODAY)
+        assert result.blocked is True and result.is_matched is False
+        assert result.effective_level == "C3"
+        assert any("SUIT-006" in r for r in result.reasons)
+        assert not any("SUIT-008" in r for r in result.reasons)
+
+
+class TestSUIT008Validity:
+    def test_364_days_valid(self):
+        assert is_assessment_valid(TODAY - timedelta(days=364), TODAY, 365) is True
+
+    def test_365_days_expired(self):
+        assert is_assessment_valid(TODAY - timedelta(days=365), TODAY, 365) is False
+
+    def test_none_evaluated_expired(self):
+        assert is_assessment_valid(None, TODAY, 365) is False
+
+    def test_expired_but_matched_still_blocked(self):
+        """C3 买 R3 等级匹配，但测评过期 → 仍阻断（双字段语义）。"""
+        customer = _customer("C3", evaluated=TODAY - timedelta(days=400))
+        result = check_core(customer, _product("R3"), today=TODAY)
+        assert result.is_matched is True and result.blocked is True
+        assert "SUIT-008" in result.reasons[-1]
+        assert "重新测评" in result.block_reason
+
+    def test_custom_valid_days(self):
+        customer = _customer("C3", evaluated=TODAY - timedelta(days=100))
+        assert is_assessment_valid(customer["risk_evaluated_at"], TODAY, 90) is False
+
+
+class TestCheckService:
+    """suitability_check 服务集成（sqlite 内存库）：落日志 + 原等级落库。"""
+
+    @pytest.fixture()
+    def repos(self):
+        engine = create_engine("sqlite:///:memory:")
+        with engine.begin() as conn:
+            conn.execute(
+                text(
+                    """
+                    CREATE TABLE core_customer (
+                        customer_id VARCHAR(64) PRIMARY KEY, display_name VARCHAR(64),
+                        age INTEGER, occupation VARCHAR(64), phone_mask VARCHAR(16),
+                        tenant_id VARCHAR(32), open_date DATE, is_active INTEGER DEFAULT 1,
+                        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                    )
+                    """
+                )
+            )
+            conn.execute(
+                text(
+                    """
+                    CREATE TABLE core_customer_risk (
+                        id INTEGER PRIMARY KEY, customer_id VARCHAR(64), risk_code VARCHAR(2),
+                        is_authoritative INTEGER DEFAULT 1, evaluated_at DATE,
+                        source VARCHAR(32) DEFAULT 'risk_questionnaire'
+                    )
+                    """
+                )
+            )
+            conn.execute(
+                text(
+                    """
+                    CREATE TABLE core_product (
+                        product_id VARCHAR(64) PRIMARY KEY, product_name VARCHAR(128),
+                        product_type VARCHAR(8), min_risk_code VARCHAR(2),
+                        industry_code VARCHAR(16), fee_rate NUMERIC(6,4),
+                        is_open INTEGER DEFAULT 1, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                    )
+                    """
+                )
+            )
+            conn.execute(
+                text(
+                    """
+                    CREATE TABLE risk_suitability_log (
+                        id INTEGER PRIMARY KEY, trace_id VARCHAR(64), customer_id VARCHAR(64),
+                        product_id VARCHAR(64), customer_risk_level VARCHAR(2),
+                        product_risk_level VARCHAR(2), is_matched INTEGER, is_blocked INTEGER,
+                        block_reason VARCHAR(512), request_ref VARCHAR(64),
+                        profile_l1_version INTEGER, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                    )
+                    """
+                )
+            )
+            conn.execute(
+                text(
+                    "INSERT INTO core_customer (customer_id, display_name, age, open_date)"
+                    " VALUES ('CUST-1', '客户·测**', 70, '2020-01-01')"
+                )
+            )
+            conn.execute(
+                text(
+                    "INSERT INTO core_customer_risk (customer_id, risk_code, evaluated_at)"
+                    " VALUES ('CUST-1', 'C5', :evaluated)"
+                ),
+                {"evaluated": TODAY - timedelta(days=90)},
+            )
+            conn.execute(
+                text(
+                    "INSERT INTO core_product (product_id, product_name, product_type, min_risk_code)"
+                    " VALUES ('PROD-R4', '测试R4产品', 'stock', 'R4')"
+                )
+            )
+        core_ro = CoreReadOnlyRepository(engine=engine)
+        risk_repo = RiskRepository(engine=engine)
+        yield core_ro, risk_repo, engine
+        engine.dispose()
+
+    def test_service_blocked_and_log_written(self, repos):
+        core_ro, risk_repo, engine = repos
+        result = suitability_check(
+            "CUST-1", "PROD-R4", core_ro=core_ro, risk_repo=risk_repo, today=TODAY
+        )
+        assert result.blocked is True and result.effective_level == "C3"
+        # 落库 customer_risk_level = 原测评等级 C5（封顶只进 reasons，PRD FR-2）
+        row = engine.connect().execute(
+            text(
+                "SELECT customer_risk_level, product_risk_level, is_matched, is_blocked"
+                " FROM risk_suitability_log"
+            )
+        ).first()
+        assert row[0] == "C5" and row[1] == "R4" and row[2] == 0 and row[3] == 1
+
+    def test_service_unknown_customer_raises(self, repos):
+        core_ro, risk_repo, _ = repos
+        with pytest.raises(LookupError):
+            suitability_check("NOPE", "PROD-R4", core_ro=core_ro, risk_repo=risk_repo, today=TODAY)