chore(prep): shared groundwork for batch-3 (admin/golden/cache)

第三批（T12 后台/T13 黄金集/T14 检索缓存）协调提交：
- kb：KbAssetService（登记/审核，tryon_ready 只能随 approved 打开）+ Asset 模型与 repo
- retrieval：list_gaps 缺口看板查询 + invalidate_cache() 钩子（T14 背后实现真缓存）
- 配置：ADMIN_TOKEN（空=后台 404）+ 检索缓存三参数
- pytest golden marker（默认跳过，CI 强制跑）+ Makefile golden + CI 防线 5c
- 测试隔离修复：test_auth_sms 加 autouse 清 sms:* 键夹具（频控计数跨运行
  累积会毒化后续运行——随机 IP 空间仅 254 且部分用例固定 device_id）

.env.example

@@ -143,3 +143,12 @@ SMS_HOURLY_LIMIT=5
 SMS_DAILY_LIMIT=10
 SMS_IP_DAILY_LIMIT=20
 SMS_DEVICE_DAILY_LIMIT=10
+
+# ── 极简运营后台（T-P2-09）────────────────────────────────────────────────
+# 空字符串 = 后台全部端点 404。令牌只进 .env / SSM，绝不进代码。
+ADMIN_TOKEN=
+
+# ── 检索结果缓存（T-P2-10）────────────────────────────────────────────────
+RETRIEVAL_CACHE_ENABLED=true
+RETRIEVAL_CACHE_TTL_SECONDS=60
+RETRIEVAL_CACHE_MAX_ENTRIES=512

.github/workflows/ci.yml

@@ -100,6 +100,12 @@ jobs:
           MEIDA_RUN_INTEGRATION: "1"
         run: uv run pytest -m integration --no-cov
 
+      # --- 防线 5c: 黄金集回归。指标退化（通过率低于阈值）直接失败（T-P2-11） ---
+      - name: Golden set regression
+        env:
+          MEIDA_RUN_INTEGRATION: "1"
+        run: uv run pytest -m golden --no-cov
+
       # --- 防线 6: 敏感信息扫描 ---
       - name: Secrets scan (detect-secrets)
         run: |

Makefile

@@ -72,6 +72,9 @@ test-unit:
 test-integration: ## 需要 make up && make migrate
 	MEIDA_RUN_INTEGRATION=1 uv run pytest -m integration
 
+golden: ## 黄金集回归（需要 PG；指标退化即失败，T-P2-11）
+	MEIDA_RUN_INTEGRATION=1 uv run pytest -m golden --no-cov
+
 check: lint typecheck imports test secrets ## 提交前一把过
 
 # ── 真实链路验证（会花钱）──────────────────────────────────────────────────

packages/kb/repo.py

@@ -15,7 +15,7 @@ from dataclasses import dataclass
 from sqlalchemy import ColumnElement, func, select, text
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from packages.shared.models import KbColor, KbEntity, KbRelation
+from packages.shared.models import KbAsset, KbColor, KbEntity, KbRelation
 
 #: 递归遍历 SQL（1~3 跳）。``path`` 记录已经过的实体 id，``rels`` 与逐跳
 #: rel_type 平行——每行结果本身就是一条完整可解释路径，不需要在 Python
@@ -166,3 +166,39 @@ async def list_colors(session: AsyncSession, *, season_code: str | None = None)
 def add_color(session: AsyncSession, color: KbColor) -> KbColor:
     session.add(color)
     return color
+
+
+async def get_asset(session: AsyncSession, asset_id: int) -> KbAsset | None:
+    return await session.get(KbAsset, asset_id)
+
+
+async def list_assets(
+    session: AsyncSession,
+    *,
+    entity_id: int | None = None,
+    tryon_ready: bool | None = None,
+    limit: int = 50,
+    offset: int = 0,
+) -> tuple[list[KbAsset], int]:
+    """素材列表（含总数）。tryon_ready 过滤走 partial 索引。"""
+    from sqlalchemy import func, select
+
+    conditions = []
+    if entity_id is not None:
+        conditions.append(KbAsset.entity_id == entity_id)
+    if tryon_ready is not None:
+        conditions.append(KbAsset.tryon_ready == tryon_ready)
+
+    base = select(KbAsset)
+    if conditions:
+        base = base.where(*conditions)
+    total = (await session.execute(select(func.count()).select_from(base.subquery()))).scalar_one()
+    rows = (
+        await session.execute(base.order_by(KbAsset.asset_id.desc()).limit(limit).offset(offset))
+    ).scalars()
+    return list(rows), int(total)
+
+
+def add_asset(session: AsyncSession, asset: KbAsset) -> KbAsset:
+    session.add(asset)
+    return asset

packages/kb/schemas.py

@@ -164,3 +164,35 @@ class ColorOut(BaseModel):
     lab_b: float
     hex: str = Field(description="#RRGGBB")
     name_cn: str
+
+
+class AssetCreate(BaseModel):
+    """素材登记（kb_asset）。cos_key 是 COS 上的对象键（真实文件经 T5 预签名
+    直传后，由运营在这里登记元数据）；spec_check 记录白底规格校验结果。"""
+
+    entity_id: str
+    asset_type: str = Field(description="display / tryon / detail 等素材类型")
+    cos_key: str = Field(min_length=1, max_length=255)
+    spec_check: dict[str, Any] = Field(default_factory=dict)
+    tryon_ready: bool = False
+    license: str | None = None
+
+
+class AssetReview(BaseModel):
+    """素材审核结论：review_status 与 tryon_ready 一起更新——
+    tryon_ready=true 是换装闭环放行的唯一开关，不允许绕过审核单独打开。"""
+
+    review_status: Literal["approved", "rejected", "pending"]
+    tryon_ready: bool = False
+
+
+class AssetOut(BaseModel):
+    asset_id: str
+    entity_id: str
+    asset_type: str
+    cos_key: str
+    spec_check: dict[str, Any]
+    tryon_ready: bool
+    review_status: str
+    license: str | None
+    created_at: datetime

packages/kb/service.py

@@ -19,6 +19,8 @@ from packages.domain.ids import next_id
 from packages.kb import repo
 from packages.kb.schemas import (
     MAX_GRAPH_HOPS,
+    AssetCreate,
+    AssetReview,
     ColorOut,
     EntityCreate,
     EntityUpdate,
@@ -30,7 +32,7 @@ from packages.kb.schemas import (
 )
 from packages.shared.errors import AppError, ErrorCode, NotFoundError
 from packages.shared.logging import get_logger
-from packages.shared.models import KbColor, KbEntity, KbRelation
+from packages.shared.models import KbAsset, KbColor, KbEntity, KbRelation
 
 logger = get_logger(__name__)
 
@@ -390,3 +392,62 @@ def _color_out(color: KbColor) -> ColorOut:
         hex=color.hex,
         name_cn=color.name_cn,
     )
+
+
+class KbAssetService:
+    """素材登记与审核（kb_asset）。与实体/图谱分开放：素材由运营流水线
+    （T-P2-07/15）高频写入，审核是唯一能把 tryon_ready 打开的入口。"""
+
+    def __init__(self, session: AsyncSession) -> None:
+        self._session = session
+
+    async def register_asset(self, payload: AssetCreate) -> KbAsset:
+        entity = await repo.get_entity(self._session, int(payload.entity_id))
+        if entity is None:
+            raise AppError(ErrorCode.NOT_FOUND, message="实体不存在，先建实体再登记素材")
+        asset = repo.add_asset(
+            self._session,
+            KbAsset(
+                asset_id=next_id(),
+                entity_id=entity.entity_id,
+                asset_type=payload.asset_type,
+                cos_key=payload.cos_key,
+                spec_check=payload.spec_check,
+                tryon_ready=False,  # 登记时恒为 False，只有审核能打开
+                review_status="pending",
+                license=payload.license,
+            ),
+        )
+        await self._session.flush()
+        return asset
+
+    async def review_asset(self, asset_id: int, payload: AssetReview) -> KbAsset:
+        asset = await repo.get_asset(self._session, asset_id)
+        if asset is None:
+            raise AppError(ErrorCode.NOT_FOUND, message="素材不存在")
+        # tryon_ready 只能随 approved 一起打开：审核是唯一放行开关
+        if payload.tryon_ready and payload.review_status != "approved":
+            raise AppError(
+                ErrorCode.BAD_REQUEST,
+                message="tryon_ready 只能在审核通过时开启",
+            )
+        asset.review_status = payload.review_status
+        asset.tryon_ready = payload.tryon_ready
+        await self._session.flush()
+        return asset
+
+    async def list_assets(
+        self,
+        *,
+        entity_id: int | None = None,
+        tryon_ready: bool | None = None,
+        limit: int = 50,
+        offset: int = 0,
+    ) -> tuple[list[KbAsset], int]:
+        return await repo.list_assets(
+            self._session,
+            entity_id=entity_id,
+            tryon_ready=tryon_ready,
+            limit=limit,
+            offset=offset,
+        )

packages/retrieval/repo.py

@@ -261,3 +261,24 @@ async def record_gap(
     else:
         gap.hit_count += 1
         gap.last_seen_at = datetime.now(UTC)
+
+
+#: 进程内缓存代数：invalidate_cache() 自增，缓存键带上它即可整体失效（T-P2-10）。
+#: 放 repo 而非 service，是因为 T14 的缓存模块会同文件引用，避免循环 import。
+INVALIDATION_VERSION = 1
+
+
+async def list_gaps(session: AsyncSession, *, limit: int = 50, offset: int = 0) -> list[RowMapping]:
+    """缺口看板查询：hit_count 降序，同频次按最近出现降序。"""
+    from sqlalchemy import select
+
+    from packages.shared.models import KbGap
+
+    stmt = (
+        select(KbGap)
+        .order_by(KbGap.hit_count.desc(), KbGap.last_seen_at.desc())
+        .limit(limit)
+        .offset(offset)
+    )
+    result = await session.execute(stmt)
+    return list(result.mappings().all())

packages/retrieval/service.py

@@ -112,3 +112,31 @@ class RetrievalService:
             logger.error("retrieval_gap_record_failed", query_len=len(query_text))
         logger.info("retrieval_gap", query_len=len(query_text), **ctx.filter_ctx())
         return SearchOutcome(results=[], degraded=True)
+
+
+class GapSummary:
+    """kb_gap 的一行摘要（缺口看板的数据形状）。V1 用普通类：字段还会随
+    看板需求演进，先不冻结成 dataclass。"""
+
+    def __init__(self, row: Any) -> None:
+        self.gap_id = row.gap_id
+        self.query_text = row.query_text
+        self.filter_ctx = row.filter_ctx
+        self.hit_count = row.hit_count
+        self.first_seen_at = row.first_seen_at
+        self.last_seen_at = row.last_seen_at
+
+
+async def list_gaps(session: AsyncSession, *, limit: int = 50, offset: int = 0) -> list[GapSummary]:
+    """缺口看板：按 hit_count 降序（最常被问到的缺口排最前，T-P2-06）。"""
+    rows = await repo.list_gaps(session, limit=limit, offset=offset)
+    return [GapSummary(r) for r in rows]
+
+
+def invalidate_cache() -> None:
+    """发布/归档知识后调用：让检索结果缓存整体失效（T-P2-10）。
+
+    第三批 T14 会在背后实现真正的进程内缓存；本桩保证调用方（运营后台的
+    发布编排）现在就能接上这个钩子，T14 合并后无需再改调用点。
+    """
+    repo.INVALIDATION_VERSION += 1

packages/shared/config.py

@@ -203,6 +203,16 @@ class Settings(BaseSettings):
     sms_ip_daily_limit: int = 20
     sms_device_daily_limit: int = 10
 
+    # --- 极简运营后台（T-P2-09）----------------------------------------------
+    # 空字符串 = 后台全部端点 404（等价于未部署）。V1 极简鉴权：共享令牌，
+    # 后续接运营账号体系时替换。令牌只进 .env / SSM，绝不进代码。
+    admin_token: str = ""
+
+    # --- 检索结果缓存（T-P2-10，T14 实现背后的配置）--------------------------
+    retrieval_cache_enabled: bool = True
+    retrieval_cache_ttl_seconds: int = 60
+    retrieval_cache_max_entries: int = 512
+
     # --- Observability -----------------------------------------------------
     otel_enabled: bool = False
     otel_service_name: str = "meida-api"

pyproject.toml

@@ -67,9 +67,14 @@ ignore = ["B008", "RUF001", "RUF002", "RUF003"]
 [tool.pytest.ini_options]
 asyncio_mode = "auto"
 testpaths = ["tests"]
-# 集成用例需要真实 PostgreSQL/Redis，默认不跑：`make test-integration` 显式开启
-addopts = "-q --strict-markers --strict-config -m 'not integration' --cov=packages --cov=apps"
-markers = ["integration: 需要真实 PostgreSQL / Redis（默认跳过）"]
+# 集成用例需要真实 PostgreSQL/Redis，默认不跑：`make test-integration` 显式开启。
+# golden 用例（黄金集回归）同样需要真实 PG：CI 单独一步强制跑（T-P2-11），
+# 本地默认跳过、`make golden` 显式开启。
+addopts = "-q --strict-markers --strict-config -m 'not integration and not golden' --cov=packages --cov=apps"
+markers = [
+    "integration: 需要真实 PostgreSQL / Redis（默认跳过）",
+    "golden: 黄金集回归（需要真实 PG；指标退化即失败，CI 强制跑）",
+]
 
 [tool.coverage.run]
 branch = true

tests/integration/test_auth_sms.py

@@ -56,6 +56,26 @@ def _rand_ip() -> str:
     return f"198.51.100.{uuid.uuid4().int % 254 + 1}"
 
 
+@pytest.fixture(autouse=True)
+async def _clean_sms_counters() -> AsyncIterator[None]:
+    """每用例前清空 sms:* 计数键。
+
+    频控计数落在常驻 Redis 里，会跨 pytest 运行累积（随机 IP 只从
+    198.51.100.0/24 取值、部分用例固定 device_id），历史运行的计数会把
+    本运行的合法发送挤过阈值。测试隔离靠自己清键，不依赖随机性。
+    """
+    if not _infra_available():
+        yield
+        return
+    from packages.shared.redis_client import get_redis
+
+    redis = await get_redis()
+    keys = [k async for k in redis.scan_iter(match="sms:*")]
+    if keys:
+        await redis.delete(*keys)
+    yield
+
+
 @pytest.fixture
 async def env() -> AsyncIterator[tuple[Any, CapturingSender]]:
     from apps.api.main import create_app