test(e2e): 运营登录只用 seed 账号；补幂等 E2E 与两处机械守卫

1) 修复一处从未暴露的 CI 缺陷

库中 operator 账号没有任何迁移种子，DemoDataSeeder 也不创建运营账号
（只统计数量），全新库唯一的账号来源是 seed_e2e_operators.sql。
而 conftest 的 admin_token 写死 username="admin" —— 那是某台开发机上
手工创建的遗留账号：本机有它所以本地一直跑绿，CI 全新库没有它 →
该 fixture 必然 401 → e2e job 从未真正通过过。
同一文件里「外加迁移 seed 的 admin」这句注释也是纯粹的虚构。

改用 seed 创建的 e2e_admin（已绑定 super_admin，语义等价）。

2) fixture_account_guard_test.py（2 用例）

机械拦住「fixture 依赖本机手工数据」：从 seed SQL 解析账号清单，
扫描 conftest 的所有登录用户名，任何一个不在 seed 内即变红；
另一条锁住推理前提（迁移里确实没有 operator 种子）。
两条都带「解析失败即失败」断言 —— 守卫自身静默失效是本项目反复踩过的坑。

3) idempotency_test.py（7 用例）

同键重放响应逐字节相同、同键重复提交只产生一条记录、body 字段幂等未回归、
不同键相互独立、同键作用域按 actor 隔离、5 线程并发同键业务恰好落库一次、
无键重复 Like 不得返回 500。

4) conftest.ApiClient.call 支持 headers 覆盖，供幂等用例携带自定义键。

tests/e2e/conftest.py

@@ -49,12 +49,16 @@ class ApiClient:
         token: Optional[str] = None,
         json: Optional[Dict[str, Any]] = None,
         params: Optional[Dict[str, Any]] = None,
+        headers: Optional[Dict[str, str]] = None,
     ) -> requests.Response:
-        headers = {"Authorization": f"Bearer {token}"} if token else {}
+        merged = {"Authorization": f"Bearer {token}"} if token else {}
+        if headers:
+            # 显式传入的头覆盖默认头（幂等键等按用例定制）
+            merged.update(headers)
         return self.session.request(
             method,
             f"{self.base_url}{path}",
-            headers=headers,
+            headers=merged,
             json=json,
             params=params,
             timeout=self.timeout,
@@ -112,15 +116,29 @@ def operator_password() -> str:
     return pwd
 
 
+ADMIN_USERNAME = "e2e_admin"
+
+
 @pytest.fixture(scope="session")
 def admin_token(api: ApiClient, operator_password: str) -> str:
+    """超级管理员 token。
+
+    必须使用 **seed 脚本创建的账号**，而不是 `admin`。
+
+    原因（本轮发现）：库中 operator 账号**没有任何迁移种子**，`DemoDataSeeder`
+    也不创建运营账号（只统计），运营账号只能由运行时服务或本 seed 脚本产生。
+    此前该 fixture 写的是 `username="admin"` —— 那是某台开发机上手工创建的遗留
+    账号，全新库（CI）里根本不存在 → 该 fixture 在 CI 上必然 401，
+    **整个 E2E job 从未真正通过过**，只是本机因为有遗留账号而看不出来。
+    `e2e_admin` 由 seed 绑定 super_admin 角色，语义等价且可自举。
+    """
     data = api.ok(
         "POST",
         "/auth/operator/login",
-        json={"username": "admin", "password": operator_password},
+        json={"username": ADMIN_USERNAME, "password": operator_password},
     )
     token = data.get("accessToken")
-    assert token, "admin 登录未返回 accessToken"
+    assert token, f"{ADMIN_USERNAME} 登录未返回 accessToken"
     return token
 
 
@@ -130,9 +148,13 @@ def operator_token(
 ) -> Callable[[str], str]:
     """按用户名取运营 token（会话内缓存）。
 
-    可用账号见 tests/e2e/sql/seed_e2e_operators.sql：
+    可用账号**仅限** tests/e2e/sql/seed_e2e_operators.sql 中创建的 6 个：
-    e2e_admin / e2e_platform / e2e_counselor / e2e_station / e2e_matchmaker / e2e_content，
+    e2e_admin / e2e_platform / e2e_counselor / e2e_station / e2e_matchmaker / e2e_content。
-    外加迁移 seed 的 admin。用法：`operator_token("e2e_counselor")`。
+    迁移里**没有任何 operator 种子**（原文"外加迁移 seed 的 admin"是错的，
+    已纠正——全新库里没有 admin 这个账号）。用法：`operator_token("e2e_counselor")`。
+
+    传入 seed 之外的用户名等于依赖某台机器上的手工数据，CI 上必然 401；
+    `fixture_account_guard_test.py` 会机械拦住这类写法。
     """
     cache: Dict[str, str] = {}
 

tests/e2e/fixture_account_guard_test.py

@@ -0,0 +1,73 @@
+"""E2E 自检守卫：fixture 只能使用 seed 脚本真正创建的运营账号。
+
+为什么需要它（本轮发现的真实缺陷）：
+
+    库中 operator 账号**没有任何迁移种子**，`DemoDataSeeder` 也不创建运营账号
+    （只统计数量）。因此全新库（CI）里唯一的运营账号来源是
+    `tests/e2e/sql/seed_e2e_operators.sql`。
+
+    而 `conftest.py` 的 `admin_token` 曾写死 `username="admin"` —— 那是某台
+    开发机上手工创建的遗留账号。本机有它，所以本地跑绿；CI 全新库没有它，
+    该 fixture 必然 401 → **整个 E2E job 从未真正通过过**。
+
+    这类"依赖本机手工数据"的写法没有任何运行期提示，只能靠机械检查拦住。
+"""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+E2E_DIR = Path(__file__).resolve().parent
+CONFTEST = E2E_DIR / "conftest.py"
+SEED_SQL = E2E_DIR / "sql" / "seed_e2e_operators.sql"
+MIGRATION_DIR = E2E_DIR.parent.parent / "backend" / "src" / "main" / "resources" / "db" / "migration"
+
+
+def _seeded_usernames() -> set[str]:
+    sql = SEED_SQL.read_text(encoding="utf-8")
+    # seed 用的是 VALUES 行：('uuid', '名称', 'username', crypt(...), 'active')
+    return set(re.findall(r"'([A-Za-z0-9_]+)',\s+crypt\(", sql))
+
+
+def _usernames_used_in_conftest() -> set[str]:
+    src = CONFTEST.read_text(encoding="utf-8")
+    used = set(re.findall(r'"username"\s*:\s*"([A-Za-z0-9_]+)"', src))
+    used |= set(re.findall(r'ADMIN_USERNAME\s*=\s*"([A-Za-z0-9_]+)"', src))
+    return used
+
+
+def test_conftest_only_logs_in_with_seeded_accounts() -> None:
+    seeded = _seeded_usernames()
+    assert seeded, f"未能从 {SEED_SQL.name} 解析出任何账号，守卫已失效"
+
+    used = _usernames_used_in_conftest()
+    assert used, f"未能从 {CONFTEST.name} 解析出任何登录用户名，守卫已失效"
+
+    unknown = sorted(used - seeded)
+    assert not unknown, (
+        "conftest 使用了 seed 脚本未创建的运营账号："
+        f"{unknown}（seed 现有：{sorted(seeded)}）。\n"
+        "全新库（CI）里这些账号不存在，对应 fixture 会 401，E2E 必然失败。"
+        "请改用 seed 中的账号，或在 seed 脚本中补上该账号。"
+    )
+
+
+def test_no_operator_account_is_seeded_by_migrations() -> None:
+    """锁住上面那条推理的前提：迁移确实没有 operator 种子。
+
+    若将来有人在迁移里加了 admin 种子，本用例会变红——那时应重新评估
+    `admin_token` 的选择（迁移种子同样是自举的，可作为合法来源），
+    而不是继续凭本机的遗留账号。
+    """
+    hits: list[str] = []
+    for sql_file in MIGRATION_DIR.glob("*.sql"):
+        text = sql_file.read_text(encoding="utf-8", errors="replace")
+        if re.search(r"INSERT\s+INTO\s+operators\b", text, re.IGNORECASE):
+            hits.append(sql_file.name)
+
+    assert not hits, (
+        f"迁移中出现了 operator 种子：{hits}。"
+        "本次 E2E fixture 的选型前提（'运营账号只能来自 seed 脚本'）已改变，"
+        "请重新评估 conftest 的 ADMIN_USERNAME 与本文档说明。"
+    )

tests/e2e/idempotency_test.py

@@ -0,0 +1,219 @@
+"""
+幂等契约 E2E（规范第八章「Idempotency Key」+ 第一百四十六章「接口幂等验收」）。
+
+背景：本轮检修发现后端**零处**读取 `Idempotency-Key` 头 —— 小程序有 26 处
+生成并发送该头，全部石沉大海；唯一真正实现幂等的是 counseling 的 body 字段。
+规范第 146 章的验收口径是「**最终不能生成重复核心数据**」。
+
+本文件验证的是**机制本身**，不是某个端点的业务规则：
+- 同一把键的重复投递 → 重放首次响应，业务代码只执行一次；
+- 不同键 / 不同主体 → 互不干扰；
+- 并发同键 → 业务只落地一次，且**绝不出现 500**。
+
+设计纪律与 conftest 一致：栈不可达时退出（码 4），不退化为 skip。
+"""
+from __future__ import annotations
+
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+from typing import Any, Dict
+
+import pytest
+
+pytestmark = pytest.mark.e2e
+
+IDEMPOTENCY_HEADER = "Idempotency-Key"
+SERVICE_TYPE = "relationship_guidance"
+
+
+def _key(scope: str) -> str:
+    """每次调用生成一把全新的键 —— 模拟「一次新的逻辑操作」。"""
+    return f"e2e-{scope}-{uuid.uuid4()}"
+
+
+def _headers(key: str) -> Dict[str, str]:
+    return {IDEMPOTENCY_HEADER: key}
+
+
+def _counseling_body() -> Dict[str, Any]:
+    return {
+        "serviceType": SERVICE_TYPE,
+        "summary": "e2e 幂等验证",
+        "requestSource": "user_self",
+    }
+
+
+def _my_services(api: Any, token: str) -> list:
+    return api.ok("GET", "/counseling/services", token=token) or []
+
+
+# ------------------------------------------------------------------ 重放语义
+
+
+def test_same_key_replays_identical_response(api, new_user):
+    """同一把键的重复投递必须**重放**首次响应，而不是重新执行。
+
+    这是幂等键最核心的语义，也是改造前完全失效的地方。
+
+    刻意用「咨询申请」而不是「喜欢」来验证：新注册用户的账号状态尚未通过
+    审核，Like 会被业务规则拒绝（`账号状态或资料未通过，无法 Like`）。
+    那是**业务前置条件**，与幂等机制无关，混进来会让用例红得没有信息量。
+    """
+    me = new_user()
+    key = _key("replay")
+    before = len(_my_services(api, me["token"]))
+
+    first = api.call("POST", "/counseling/services", token=me["token"],
+                     json=_counseling_body(), headers=_headers(key))
+    second = api.call("POST", "/counseling/services", token=me["token"],
+                      json=_counseling_body(), headers=_headers(key))
+
+    assert first.status_code == 200, (
+        f"首次提交失败：HTTP {first.status_code} {first.text[:300]}"
+    )
+    assert second.status_code == 200, (
+        f"重复请求未被重放（HTTP {second.status_code}）：{second.text[:300]}"
+    )
+    assert first.json() == second.json(), (
+        "同一幂等键的两次请求必须返回完全一致的响应，"
+        f"实际不一致：\nfirst={first.json()}\nsecond={second.json()}"
+    )
+    assert len(_my_services(api, me["token"])) == before + 1, (
+        "重放不应额外落地记录"
+    )
+
+
+def test_repeated_submission_with_same_key_creates_single_record(api, new_user):
+    """弱网重发场景：同一把键重复提交咨询申请，只允许落地一条记录。"""
+    me = new_user()
+    key = _key("counseling")
+    before = len(_my_services(api, me["token"]))
+
+    first = api.call("POST", "/counseling/services", token=me["token"],
+                     json=_counseling_body(), headers=_headers(key))
+    second = api.call("POST", "/counseling/services", token=me["token"],
+                      json=_counseling_body(), headers=_headers(key))
+
+    assert first.status_code == 200, f"首次提交失败：{first.text[:300]}"
+    assert second.status_code == 200, f"重复提交失败：{second.text[:300]}"
+
+    after = _my_services(api, me["token"])
+    assert len(after) == before + 1, (
+        f"同一幂等键重复提交产生了 {len(after) - before} 条申请，期望 1 条"
+    )
+
+
+def test_body_idempotency_key_still_prevents_duplicates(api, new_user):
+    """counseling 原有的 body 字段幂等（V039 `uk_counseling_idem`）不得因新机制回归。"""
+    me = new_user()
+    body = _counseling_body()
+    body["idempotencyKey"] = _key("counseling-body")
+    before = len(_my_services(api, me["token"]))
+
+    api.ok("POST", "/counseling/services", token=me["token"], json=body)
+    api.ok("POST", "/counseling/services", token=me["token"], json=body)
+
+    after = _my_services(api, me["token"])
+    assert len(after) == before + 1, (
+        f"body 幂等键失效：重复提交产生了 {len(after) - before} 条申请"
+    )
+
+
+# ------------------------------------------------------------------ 作用域隔离
+
+
+def test_different_keys_are_independent_operations(api, new_user):
+    """不同键 = 两次独立操作，都应被受理（幂等不能变成"同端点只准提交一次"）。"""
+    me = new_user()
+    before = len(_my_services(api, me["token"]))
+
+    api.ok("POST", "/counseling/services", token=me["token"],
+           json=_counseling_body(), headers=_headers(_key("scope-a")))
+    api.ok("POST", "/counseling/services", token=me["token"],
+           json=_counseling_body(), headers=_headers(_key("scope-b")))
+
+    after = _my_services(api, me["token"])
+    assert len(after) == before + 2, (
+        f"不同幂等键的两次提交应各成一条，实际新增 {len(after) - before} 条"
+    )
+
+
+def test_same_key_is_scoped_per_actor(api, new_user):
+    """同一把键被两个用户使用：各算各的操作，绝不能把 A 的响应重放给 B。
+
+    键的作用域是「主体 + 端点 + 键」。若只按键去重，B 会拿到 A 的申请结果，
+    这既是数据泄露也是功能错误。
+    """
+    user_a = new_user()
+    user_b = new_user()
+    key = _key("shared-key")
+    before_a = len(_my_services(api, user_a["token"]))
+    before_b = len(_my_services(api, user_b["token"]))
+
+    resp_a = api.call("POST", "/counseling/services", token=user_a["token"],
+                      json=_counseling_body(), headers=_headers(key))
+    resp_b = api.call("POST", "/counseling/services", token=user_b["token"],
+                      json=_counseling_body(), headers=_headers(key))
+
+    assert resp_a.status_code == 200, f"A 提交失败：{resp_a.text[:300]}"
+    assert resp_b.status_code == 200, (
+        f"B 使用与 A 相同的键却被拒绝/重放（HTTP {resp_b.status_code}）：{resp_b.text[:300]}"
+    )
+    assert resp_a.json()["data"]["id"] != resp_b.json()["data"]["id"], (
+        "两个用户使用同一把键时拿到了同一条记录 —— 幂等作用域未按主体隔离"
+    )
+    assert len(_my_services(api, user_a["token"])) == before_a + 1
+    assert len(_my_services(api, user_b["token"])) == before_b + 1
+
+
+# ------------------------------------------------------------------ 并发兜底
+
+
+def test_concurrent_same_key_lands_business_exactly_once(api, new_user):
+    """并发同键（连点两次的真实形态）：业务只落地一次，且不得出现 500。
+
+    这是占位机制存在的理由 —— 只在结束时才记录的话，两个并发请求会
+    **都跑完业务**再发现重复，副作用已经发生。
+    """
+    me = new_user()
+    key = _key("concurrent")
+    before = len(_my_services(api, me["token"]))
+
+    def send(_: int):
+        return api.call("POST", "/counseling/services", token=me["token"],
+                        json=_counseling_body(), headers=_headers(key))
+
+    with ThreadPoolExecutor(max_workers=5) as pool:
+        responses = list(pool.map(send, range(5)))
+
+    codes = [r.status_code for r in responses]
+    assert 500 not in codes, (
+        f"并发重复提交出现 500（应为 200 重放或 409 处理中）：{codes}"
+        f"\n{responses[codes.index(500)].text[:400]}"
+    )
+    assert all(code in (200, 409) for code in codes), f"出现非预期状态码：{codes}"
+
+    after = _my_services(api, me["token"])
+    assert len(after) == before + 1, (
+        f"并发同键落地了 {len(after) - before} 条记录，期望恰好 1 条"
+    )
+
+
+# ------------------------------------------------------------------ 无键路径不退化
+
+
+def test_duplicate_like_without_key_is_not_500(api, new_user):
+    """不带幂等键的重复提交仍不得 500。
+
+    这条守的是另一条防线：业务层的 SELECT-then-INSERT 与 DB 唯一约束
+    （`uk_user_relation`）。并发撞约束时应是 409，不是服务器错误。
+    """
+    me = new_user()
+    target = new_user()
+    payload = {"targetUserId": target["user_id"]}
+
+    first = api.call("POST", "/likes", token=me["token"], json=payload)
+    second = api.call("POST", "/likes", token=me["token"], json=payload)
+
+    assert first.status_code != 500, f"首次喜欢 500：{first.text[:300]}"
+    assert second.status_code != 500, f"重复喜欢 500：{second.text[:300]}"