fix(scoring): 口径常量下沉 core + repositories 逆向依赖门禁 + 前端口径镜像门禁 (#113 #112)

.github/workflows/ci.yml

@@ -123,6 +123,22 @@ jobs:
       - name: Check AGENTS.md references
         run: python scripts/check_agent_docs.py ../AGENTS.md
 
+  # ── 前后端评分口径一致性（#112）──
+  # 前端 recommendation.ts 镜像的两个门槛数字（82/45）与
+  # app/core/scoring_caliber.py CONFIG 对比；#110 的学费不允许再交一次。
+  scoring-mirror:
+    runs-on: ubuntu-latest
+    defaults:
+      run:
+        working-directory: backend
+    steps:
+      - uses: actions/checkout@v5
+      - uses: actions/setup-python@v6
+        with:
+          python-version: "3.12"
+      - name: Check scoring caliber mirror
+        run: python scripts/check_scoring_mirror.py
+
   # ── 依赖安全扫描 ──
   security-scan:
     runs-on: ubuntu-latest

AGENTS.md

@@ -60,14 +60,19 @@ api/v1/ ──► services/ ──► repositories/ ──► models/ ──►
 `IntegrityError` 等异常类。
 
 **机器强制的边界**：`make layering`（等价于 `cd backend && uv run python
-scripts/check_layering.py`，CI 作业 `backend-layering`）只作用于 `app/api/v1/`，且只认两条：
-禁止 `import sqlalchemy`（`sqlalchemy.ext.asyncio` / `sqlalchemy.exc` 除外），禁止直接写
-`select/insert/update/delete` 构造与 `db.execute/db.add/db.scalars/db.scalar`。上表其余禁止项
-它查不出——`from app.models import <ORMModel>` 与 api 层合法要用的 Enum、依赖注入同模块，
-自动判定会误报，只能人工看。`app/mcp/` 是与 `api/v1` 平级的入口层，不在机器检查范围内，
-按上表 api 行的纪律人工把关（当前只调 services 与 `DailyReportRepository`；OAuth
-provider 的业务编排在 `app/services/mcp_oauth_service.py`，同意页 REST 端点在
-`app/api/v1/mcp_oauth.py`）。
+scripts/check_layering.py`，CI 作业 `backend-layering`）作用于两个目录，共三条规则。
+`app/api/v1/` 认两条：禁止 `import sqlalchemy`（`sqlalchemy.ext.asyncio` /
+`sqlalchemy.exc` 除外），禁止直接写 `select/insert/update/delete` 构造与
+`db.execute/db.add/db.scalars/db.scalar`。`app/repositories/` 认一条（#113）：
+禁止 import `app.services.*`（函数内局部 import 同样抓）；唯一存量豁免是
+`analysis_repo.py` 的两个评分编排 import（评分上移重构单 #122 清空，
+allowlist 只减不增）。跨层共享的口径常量（`scoring_caliber.py`、
+`tag_normalization.py`）已落 `app/core/`，repositories 从 core 取用是合法向下依赖。
+上表其余禁止项机器查不出——`from app.models import <ORMModel>` 与 api 层合法要用的
+Enum、依赖注入同模块，自动判定会误报，只能人工看。`app/mcp/` 是与 `api/v1` 平级的
+入口层，不在机器检查范围内，按上表 api 行的纪律人工把关（当前只调 services 与
+`DailyReportRepository`；OAuth provider 的业务编排在
+`app/services/mcp_oauth_service.py`，同意页 REST 端点在 `app/api/v1/mcp_oauth.py`）。
 
 ### 2.3 迁移与高风险变更
 
@@ -105,10 +110,11 @@ make test-backend                         # 或直接跑全量
 - Backend tests: `cd backend && uv run python -m pytest <relevant-tests> -q` —— `-k` 是 pytest 的**全局**过滤器，`pytest a.py b.py -k x` 会同时筛两个文件，不限于 a.py；要只筛某个文件就分开跑（依赖由 uv 管理：`uv sync` 安装，事实源为 pyproject.toml + uv.lock）
 - Shell scripts: `bash -n <script>`
 - Layering: `make layering`
+- Scoring mirror: `cd backend && uv run python scripts/check_scoring_mirror.py`（前端镜像口径对比 CONFIG，CI 作业 `scoring-mirror`）
 - AGENTS.md 引用: `cd backend && uv run python scripts/check_agent_docs.py`（校验本文引用的路径与 make 目标，CI 作业 `agent-docs`）
 - Frontend type check: `cd frontend && npx tsc --noEmit`
 - Frontend tests: `cd frontend && npm run test:coverage`（含覆盖率门禁，对应 CI `frontend-tests`）
-- Full quality gate: `make lint`（ruff + 分层检查 + 前端类型检查）
+- Full quality gate: `make lint`（ruff + 分层检查 + 口径一致性 + 前端类型检查）
 
 ## 四、怎么交
 
@@ -161,7 +167,7 @@ docs: 跑测前必读补上 OAuth 隔离测试为什么必须 python -m
 
 ### 4.4 Verifier 契约
 
-CI 六项检查（types / tests / lint / layering / agent-docs / security-scan）是机器
+CI 七项检查（types / tests / lint / layering / scoring-mirror / agent-docs / security-scan）是机器
 Verifier，全绿是合并前提；预存红必须显式处置并留记录。PR 描述须附 Worker 证据（本地
 验证命令 + 结果，见 PR 模板 Verification 段）。承载行为变更的 PR 还需在模板 Verifier
 verdict 段落记录独立复核结论（复核者不得是同一实现过程）。

Makefile

@@ -9,12 +9,12 @@
 #   make setup          安装后端 + 前端依赖
 #   make dev            docker compose 起全栈
 #   make test           跑后端 + 前端测试
-#   make lint           完整质量门（ruff + 分层 + 前端类型检查）
+#   make lint           完整质量门（ruff + 分层 + 口径一致 + 前端类型检查）
 # =============================================================================
 
 .DEFAULT_GOAL := help
 .PHONY: help setup dev dev-db test test-backend test-frontend \
-        lint lint-backend lint-frontend layering backup
+        lint lint-backend lint-frontend layering scoring-mirror backup
 
 help:  ## 列出所有可用目标
 	@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) \
@@ -38,14 +38,17 @@ test-backend:  ## 后端全量 pytest（自动起一次性 PG 容器，跑完即
 test-frontend:  ## 前端 vitest + 覆盖率门禁（对应 CI frontend-tests）
 	cd frontend && npm run test:coverage
 
-lint: lint-backend layering lint-frontend  ## 完整质量门（ruff + 分层 + tsc）
+lint: lint-backend layering scoring-mirror lint-frontend  ## 完整质量门（ruff + 分层 + 口径一致 + tsc）
 
 lint-backend:  ## 后端 ruff 全量检查 + 格式检查（手动质量门）
 	cd backend && uv run ruff check . && uv run ruff format --check .
 
-layering:  ## 分层纪律检查：api/v1 禁止直接 ORM 查询 / 禁用 import
+layering:  ## 分层纪律检查：api/v1 禁 ORM 查询；repositories 禁 services 逆向 import
 	cd backend && uv run python scripts/check_layering.py
 
+scoring-mirror:  ## 前后端评分口径一致性（前端镜像 82/45 对比 CONFIG，#112）
+	cd backend && uv run python scripts/check_scoring_mirror.py
+
 lint-frontend:  ## 前端类型检查（AGENTS.md 首选，比 npm run lint 稳）
 	cd frontend && npx tsc --noEmit
 

backend/app/api/v1/contents.py

@@ -147,9 +147,7 @@ async def list_contents(
     tag: str | None = Query(None, description="按标签筛选（规范化键，大小写不敏感；命中 content 或任一分析的标签）"),
     include_trend_sources: bool = Query(False, description="Include榜单/趋势源 such as DouyinHot"),
     hours: int | None = Query(None, description="Time range in hours, e.g. 24, 48, 168"),
-    sort_by: str = Query(
-        "created_at", pattern=r"^(created_at|published_at|crawled_at|curation_score)$"
-    ),
+    sort_by: str = Query("created_at", pattern=r"^(created_at|published_at|crawled_at|curation_score)$"),
     sort_order: str = Query("desc", pattern=r"^(asc|desc)$"),
     admin_view: bool = Query(False, description="Return management fields; admin only"),
     current_user: User | None = Depends(get_optional_current_user),
@@ -416,9 +414,9 @@ async def content_tag_facets(
     from collections import Counter
     from datetime import timedelta
 
+    from app.core.tag_normalization import normalize_tag_list
     from app.repositories.ignored_repo import IgnoredRepo
     from app.services.json_cache import get_cached_json, set_cached_json
-    from app.services.tag_normalization import normalize_tag_list
 
     scope_q = (q or "").strip()
     cache_key = (

backend/app/repositories/content_repo.py

@@ -594,8 +594,8 @@ async def list_paginated_with_analyses(
         )
 
         # ── 服务端推荐等级 / 标签筛选 ────────────────────────────────
+        from app.core.tag_normalization import normalize_tag_key
         from app.models.analysis import AiAnalysis
-        from app.services.tag_normalization import normalize_tag_key
 
         # 两者口径与前端展示保持一致：等级取最新一条分析；标签命中
         # content.tags 或任一分析的 tags（对应前端 getItemTags 的合并）。
@@ -866,8 +866,8 @@ async def list_for_today_picks(
         (``owner_user_id IS NULL``) or content owned by that user. ``None``
         means no visibility filter (batch/internal callers).
         """
+        from app.core.scoring_caliber import CONFIG as SCORING_CONFIG
         from app.models.analysis import AiAnalysis
-        from app.services.scoring_engine import CONFIG as SCORING_CONFIG
 
         cutoff = naive_utc_now() - timedelta(hours=hours)
         risk_threshold = float(SCORING_CONFIG["risk_threshold"])
@@ -908,8 +908,8 @@ async def list_for_report_window(
         口径与 today-picks / DuckDB query_today_picks 对齐：剔除事件附属内容
         与已忽略内容 (``exclude_ids``)。
         """
+        from app.core.scoring_caliber import CONFIG as SCORING_CONFIG
         from app.models.analysis import AiAnalysis
-        from app.services.scoring_engine import CONFIG as SCORING_CONFIG
 
         risk_threshold = float(SCORING_CONFIG["risk_threshold"])
         latest_analysis_id = self._latest_analysis_id_subquery(AiAnalysis)

backend/app/services/analysis.py

@@ -324,8 +324,8 @@ def _build_analysis_record(
     标签与推荐等级在此收口（幂等）：lite 级联与本地回退路径不经过
     _normalize_analysis_result，统一在构造处保证存储口径一致。
     """
+    from app.core.tag_normalization import normalize_tag_list
     from app.services.recommendation_level import classify_recommendation_level
-    from app.services.tag_normalization import normalize_tag_list
 
     scores = result.get("scores", {})
     curation = result.get("curation", {})

backend/app/services/analysis_normalize.py

@@ -20,8 +20,8 @@
 import json
 from typing import Any
 
+from app.core.tag_normalization import normalize_tag_list
 from app.services.recommendation_level import classify_recommendation_level
-from app.services.tag_normalization import normalize_tag_list
 
 
 def _detect_lang(title: str, content: str) -> str:

backend/app/services/content_pipeline.py

@@ -19,6 +19,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.config import settings
+from app.core.tag_normalization import normalize_tag_list
 from app.models.content import ContentItem, ContentStatus
 from app.models.source import Source, SourceStatus, SourceType
 
@@ -31,7 +32,6 @@
 from app.services.llm_pre_filter import apply_pre_filter
 from app.services.scraper_http import build_scraper_client_kwargs
 from app.services.scrapers import get_scraper_cls
-from app.services.tag_normalization import normalize_tag_list
 from app.utils.url_safety import UnsafeUrlError, ensure_public_hostname
 
 logger = logging.getLogger(__name__)

backend/app/services/llm/prompts/analysis.py

@@ -4,11 +4,11 @@
 Used by:
     - app.services.analysis.analyze_content
 
-评分口径(加权/门槛/风险扣分)一律从 app.services.scoring_caliber 渲染,
+评分口径(加权/门槛/风险扣分)一律从 app.core.scoring_caliber 渲染,
 本文件不得手写数字——tests/test_scoring_caliber.py 锁死漂移(#110)。
 """
 
-from app.services.scoring_caliber import (
+from app.core.scoring_caliber import (
     curation_threshold_text,
     curation_weights_text,
     dimension_anchor_section,

backend/app/services/scoring_engine.py

@@ -19,16 +19,16 @@
 import math
 from datetime import UTC, datetime
 
+from app.core.scoring_caliber import CONFIG
 from app.schemas.scoring import (
     ScoreBreakdownResponse,
     ScoreResultItem,
     ScoringRequestItem,
     ScoringResponse,
 )
-from app.services.scoring_caliber import CONFIG
 
 # ── Tunable configuration ────────────────────────────────────────────
-# 唯一声明处已迁至 app/services/scoring_caliber.py(#110):prompt 教学与
+# 唯一声明处为 app/core/scoring_caliber.py(#110,#113):prompt 教学与
 # picks 查询共享同一份口径,tests/test_scoring_caliber.py 锁死漂移。
 
 

backend/scripts/check_layering.py

@@ -1,21 +1,30 @@
 #!/usr/bin/env python3
 """分层纪律的机器强制检查（对应 AGENTS.md「Layering Discipline」）。
 
-只机器化 AGENTS.md 里*符号级、可无歧义判定*的两条禁止项，作用于 `app/api/v1/`：
+机器化 AGENTS.md 里*符号级、可无歧义判定*的禁止项，作用于两个目录：
+
+**app/api/v1/**（两条）：
 
 1. 禁止 import sqlalchemy —— 例外：`sqlalchemy.ext.asyncio`（AsyncSession 类型注解）
    与 `sqlalchemy.exc`（IntegrityError / OperationalError 等异常类，供 try/except）。
 2. 禁止在 api 层直接写 ORM 查询：`select()/insert()/update()/delete()` 构造，
    以及 `db.execute(...) / db.add(...) / db.scalars(...) / db.scalar(...)`。
    这些必须下沉到 `repositories/`。
 
+**app/repositories/**（一条，#113）：
+
+3. 禁止 import `app.services.*` —— 依赖方向 repositories→services 逆向
+   （函数内局部 import 同样抓：#113 的违例全部藏在函数体内）。
+   已知存量豁免走 `REPO_SERVICES_IMPORT_ALLOWLIST`（按文件+模块粒度，
+   引用重构 issue，棘轮式收缩，只减不增）。
+
 **不检查**的项（因为是符号级且和允许项同模块，自动判定会大量误报）：
 `from app.models import <ORMModel>` 与 `<ORMModel>` 混在 Enum / User 依赖注入里的情况，
 仍靠 AGENTS.md 2.2 分层表「禁止」列的人工评审兜底。
 
 用法：
-    python scripts/check_layering.py            # 检查默认目录
-    python scripts/check_layering.py app/api/v1 # 指定目录
+    python scripts/check_layering.py            # 检查 api/v1 + repositories
+    python scripts/check_layering.py app/api/v1 # 只检查指定目录（api 规则）
 
 命中违规时打印文件:行 + 说明并以退出码 1 结束，供 CI 阻断。
 """
@@ -28,11 +37,20 @@
 
 # 检查目标目录（相对 backend/）。
 API_DIR = Path("app/api/v1")
+REPO_DIR = Path("app/repositories")
 
 # 刻意豁免的文件名（分层的合法低层边界）。
 # _db_write.py 不再需要豁免——SQLite PRAGMA 调用已随 SQLite 分支移除。
 ALLOWLIST: set[str] = set()
 
+# repositories→services 禁令的存量豁免（#113 收口时遗留，#122 重构后清空）。
+# 键 = 文件名，值 = 允许 import 的 app.services 模块前缀集合；
+# 新增条目必须附带 issue 引号，且只允许删除、不允许新增（棘轮）。
+REPO_SERVICES_IMPORT_ALLOWLIST: dict[str, set[str]] = {
+    # analysis_repo 评分编排上移 services 的重构单：#122
+    "analysis_repo.py": {"app.services.feedback_signal", "app.services.scoring_engine"},
+}
+
 # 允许的 sqlalchemy 子模块前缀（值对象 / 类型注解 / 异常类）。
 ALLOWED_SQLALCHEMY_PREFIXES = ("sqlalchemy.ext.asyncio", "sqlalchemy.exc")
 
@@ -95,16 +113,54 @@ def check_file(path: Path) -> list[str]:
     return violations
 
 
+def check_repo_services_imports(path: Path) -> list[str]:
+    """repositories 层文件：禁止 import app.services.*（allowlist 内的前缀除外）。
+
+    ast.walk 覆盖模块级与函数体内的 import——#113 的违例全部藏在函数体里，
+    只查模块级会漏。
+    """
+    allowed_prefixes = REPO_SERVICES_IMPORT_ALLOWLIST.get(path.name, set())
+    violations: list[str] = []
+    tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+
+    def _violation(lineno: int, desc: str) -> None:
+        if any(desc == prefix or desc.startswith(prefix + ".") for prefix in allowed_prefixes):
+            return
+        violations.append(
+            f"{path}:{lineno}: 禁止 `{desc}`——repositories 层不得 import services"
+            f"（业务编排属于 service，口径常量在 app/core/；存量豁免只减不增）"
+        )
+
+    for node in ast.walk(tree):
+        if isinstance(node, ast.ImportFrom):
+            module = node.module or ""
+            if module == "app.services" or module.startswith("app.services."):
+                _violation(node.lineno, module)
+        elif isinstance(node, ast.Import):
+            for alias in node.names:
+                if alias.name == "app.services" or alias.name.startswith("app.services."):
+                    _violation(node.lineno, alias.name)
+
+    return violations
+
+
 def main(argv: list[str]) -> int:
-    target = Path(argv[1]) if len(argv) > 1 else API_DIR
-    if not target.exists():
-        print(f"ERROR: 目录不存在: {target}（请在 backend/ 下运行）", file=sys.stderr)
-        return 2
+    # legacy 单目录模式只跑 api 规则（保持既有用法兼容）；默认双目录全查。
+    targets = [(Path(argv[1]), "api")] if len(argv) > 1 else [(API_DIR, "api"), (REPO_DIR, "repo")]
 
     all_violations: list[str] = []
-    files = sorted(p for p in target.glob("*.py") if p.name not in ALLOWLIST)
-    for path in files:
-        all_violations.extend(check_file(path))
+    file_count = 0
+    for target, kind in targets:
+        if not target.exists():
+            print(f"ERROR: 目录不存在: {target}（请在 backend/ 下运行）", file=sys.stderr)
+            return 2
+        files = sorted(p for p in target.glob("*.py") if p.name not in ALLOWLIST)
+        file_count += len(files)
+        for path in files:
+            if kind == "api":
+                all_violations.extend(check_file(path))
+            else:
+                all_violations.extend(check_repo_services_imports(path))
 
     if all_violations:
         print("分层违规（详见 AGENTS.md「Layering Discipline」）：", file=sys.stderr)
@@ -116,7 +172,8 @@ def main(argv: list[str]) -> int:
         )
         return 1
 
-    print(f"分层检查通过：{len(files)} 个 api/v1 文件无直接 ORM 查询 / 禁用 import。")
+    scopes = " + ".join(str(t) for t, _ in targets)
+    print(f"分层检查通过：{scopes} 共 {file_count} 个文件，无禁用 import / 逆向依赖。")
     return 0