记忆召回补一道能力码闸门：归属关系不等于授权

上一版把员工召回改成"按 sys_customer_assignment 归属"时，只看了**归属关系**，
没校验**能力码** —— 这留下一个我自己带出来的口子：

- 平台读他人画像/记忆的正式路径（customer_profile_service、public_platform_service）
  校验的是 `memory:read:customer`（sys_permission id=9010，data_scope='own_customers'）
  + `customer_ids`；
- 而召回是**另一条**把客户记忆送进模型上下文的路径，只按归属放行，等于让
  "有归属行但没有该权限"的账号（例如 operator：只有 financial:nl2sql:read
  + offsite:write）凭空获得读他人客户记忆的能力 —— 它上一版之前是读不到的。

现改为**两个条件都满足才放行**（缺一即失败关闭，并在日志里分开点名"缺能力码"
与"归属未维护"，避免运维在错误的表上找问题）：
  ① `memory:read:customer` 在 context.permissions 里；
  ② 客户落在 context.customer_ids（sys_customer_assignment 生效行）内。

演示账号不受影响：9002(risk_operator)、9020(advisor)、9003(admin) 三个角色都持有该权限码，
真实身份链路复验 9020 仍是"修复前 0 条 → 修复后 2 条"。

测试：tests/unit/core/test_memory_scope.py +3 条（缺能力码读不到／能力码与归属是"与"关系／
现有用例补上能力码），tests/unit/service/test_agent_governance.py 同步。
`pytest tests/unit tests/contract` → 1448 passed, 2 skipped, 1 failed
（唯一失败仍是组员正在改的投顾页面）。

app/core/memory_scope.py

@@ -21,10 +21,18 @@
 
 | 身份 | 可读范围 |
 |---|---|
-| **客户**（`customer` / `authenticated_user`） | **只有自己**（自身 `user_id` 当客户号）；别家客户一律不可读 |
-| **员工**（风控／投顾／运营／管理员／system） | 只有 `sys_customer_assignment` 里**分配给自己**的客户；归属未维护 ⇒ 读不到（失败关闭） |
+| **客户**（`customer` / `authenticated_user`） | **只有自己**；别家客户一律不可读 |
+| **员工** | ① 有能力码 `memory:read:customer`，**且** ② 客户已分配给自己 |
 | **访客** | 无（调用方在上游已拦） |
 
+"员工"指风控／投顾／运营／管理员／system 等一切非客户身份；上面那两条
+**缺一即读不到**（失败关闭）。
+
+**为什么要两个条件**：归属关系回答"谁负责谁"，能力码回答"能不能读他人客户数据"。
+平台读他人画像/记忆的正式路径（`customer_profile_service`、`public_platform_service`）
+校验的是后者（`memory:read:customer` + `own_customers` 范围）。召回是**另一条**把客户记忆
+送进模型上下文的路径，只按归属放行会让"有归属行但无此权限"的账号凭空获得该能力。
+
 `sys_customer_assignment` 由 `IdentityRepository.load_context()` 读入 `context.customer_ids`，
 即本模块的输入。**归属未维护不是故障，是"没有授权"** —— 所以日志必须点名它，
 让运维知道该去维护分配表，而不是去查"记忆是不是坏了"。
@@ -39,6 +47,15 @@ logger = logging.getLogger(__name__)
 #: 认定为"客户身份"的角色码。与 `app/worker/runtime.py` 的画像候选判定同一口径。
 CUSTOMER_ROLES: frozenset[str] = frozenset({"customer", "authenticated_user"})
 
+#: 跨客户读记忆所需的**能力码**（`sys_permission` id=9010，`data_scope='own_customers'`）。
+#:
+#: 为什么召回也要看它：`customer_profile_service` / `public_platform_service` 读他人画像与记忆
+#: 时校验的正是这个码 + `own_customers` 范围 + `customer_ids`。召回是**另一条**把客户记忆
+#: 送进模型上下文的路径，**只按归属关系放行是不够的** —— 归属是"谁负责谁"，能力码才是
+#: "能不能读他人客户数据"。两者都满足才放行，否则一个只有归属行、没有该权限的账号
+#: （例如 `operator`：只有 `financial:nl2sql:read` + `offsite:write`）会凭空获得读他人记忆的能力。
+REQUIRED_EMPLOYEE_PERMISSION = "memory:read:customer"
+
 #: 单次运行最多召回多少个归属客户。
 #:
 #: 为什么要有上限：员工可能有成百上千个归属客户，逐个召回会变成 N 次库查询 + N 次
@@ -70,6 +87,15 @@ def customer_memory_scope(context: RequestContext) -> tuple[int, ...]:
             return ()
         return (own,) if own > 0 else ()
 
+    # 能力码先于归属判定：两者是**与**关系。只在有权限时看归属，避免"有归属行就放行"。
+    if REQUIRED_EMPLOYEE_PERMISSION not in context.permissions:
+        logger.warning(
+            "记忆范围为空：身份 roles=%s 缺少 %s 能力（读他人客户记忆的授权码），"
+            "失败关闭。归属关系只解决「谁负责谁」，不构成读他人客户数据的授权",
+            list(context.roles), REQUIRED_EMPLOYEE_PERMISSION,
+        )
+        return ()
+
     parsed: set[int] = set()
     for raw in context.customer_ids:
         try:

app/service/agent/governance.py

@@ -17,6 +17,7 @@ from app.core.contracts import (
 from app.core.errors import ForbiddenAgentError, RecoverableAgentError
 from app.core.memory_scope import (
     RECALL_ITEM_LIMIT,
+    REQUIRED_EMPLOYEE_PERMISSION,
     customer_memory_scope,
     is_customer_identity,
     memory_customer_in_scope,
@@ -208,6 +209,14 @@ class PlatformGovernance:
                 context.user_id,
             )
             return
+        # 员工身份范围为空有两种原因，必须分开讲清楚，否则运维会在错误的表上找问题。
+        if REQUIRED_EMPLOYEE_PERMISSION not in context.permissions:
+            logger.warning(
+                "memory recall skipped: 员工身份 roles=%s 缺少 %s 能力 ⇒ 无可读客户"
+                "（失败关闭）。这不是记忆故障；要读他人客户记忆需先具备该权限码",
+                list(context.roles), REQUIRED_EMPLOYEE_PERMISSION,
+            )
+            return
         logger.warning(
             "memory recall skipped: 员工身份 roles=%s 在 sys_customer_assignment 里"
             "没有生效的归属客户 ⇒ 无可读客户（失败关闭）。这不是记忆故障；"

tests/unit/core/test_memory_scope.py

@@ -15,11 +15,15 @@ from app.core.contracts import RequestContext
 from app.core.memory_scope import (
     CUSTOMER_ROLES,
     MAX_RECALL_CUSTOMERS,
+    REQUIRED_EMPLOYEE_PERMISSION,
     customer_memory_scope,
     is_customer_identity,
     memory_customer_in_scope,
 )
 
+#: 员工读他人客户记忆的能力码；模拟 `IdentityRepository.load_context` 放进去的权限集合。
+EMPLOYEE_PERMISSIONS = (REQUIRED_EMPLOYEE_PERMISSION,)
+
 
 def _context(**overrides: object) -> RequestContext:
     base: dict[str, object] = {"user_id": "9001", "trace_id": "t"}
@@ -37,7 +41,7 @@ def test_customer_reads_only_self() -> None:
 
 def test_employee_without_assignment_reads_nothing() -> None:
     """员工号**不得**被当成客户号 —— 这正是"越权陷阱"与"恒空"的共同来源。"""
-    context = _context(user_id="9020", roles=("advisor",))
+    context = _context(user_id="9020", roles=("advisor",), permissions=EMPLOYEE_PERMISSIONS)
 
     assert is_customer_identity(context) is False
     assert customer_memory_scope(context) == ()
@@ -46,8 +50,38 @@ def test_employee_without_assignment_reads_nothing() -> None:
     assert 9020 not in customer_memory_scope(context)
 
 
+def test_employee_without_the_capability_reads_nothing() -> None:
+    """**归属关系不等于授权**：有归属行但没有 `memory:read:customer` 能力码，仍读不到。
+
+    这条守的是"召回绕开 RBAC"的口子 —— `operator` 这类只有
+    `financial:nl2sql:read` + `offsite:write` 的账号若有归属行，不能因此获得读他人记忆的能力。
+    """
+    context = _context(
+        user_id="9005", roles=("operator",), customer_ids=("9001",), permissions=("offsite:write",),
+    )
+
+    assert customer_memory_scope(context) == ()
+
+
+def test_employee_needs_both_capability_and_assignment() -> None:
+    """能力码与归属是**与**关系：缺任一条都读不到，两条齐了才给出范围。"""
+    only_capability = _context(
+        user_id="9020", roles=("advisor",), permissions=EMPLOYEE_PERMISSIONS,
+    )
+    both = _context(
+        user_id="9020", roles=("advisor",),
+        customer_ids=("9001",), permissions=EMPLOYEE_PERMISSIONS,
+    )
+
+    assert customer_memory_scope(only_capability) == ()
+    assert customer_memory_scope(both) == (9001,)
+
+
 def test_employee_reads_assigned_customers_in_deterministic_order() -> None:
-    context = _context(user_id="9020", roles=("advisor",), customer_ids=("9103", "9001", "9102"))
+    context = _context(
+        user_id="9020", roles=("advisor",), customer_ids=("9103", "9001", "9102"),
+        permissions=EMPLOYEE_PERMISSIONS,
+    )
 
     # 升序：同一身份每次得到同一批次，不随数据库返回顺序漂移。
     assert customer_memory_scope(context) == (9001, 9102, 9103)
@@ -55,7 +89,10 @@ def test_employee_reads_assigned_customers_in_deterministic_order() -> None:
 
 def test_scope_is_truncated_by_ascending_customer_id() -> None:
     many = tuple(str(10000 + index) for index in range(MAX_RECALL_CUSTOMERS + 5))
-    context = _context(user_id="9002", roles=("risk_operator",), customer_ids=many)
+    context = _context(
+        user_id="9002", roles=("risk_operator",), customer_ids=many,
+        permissions=EMPLOYEE_PERMISSIONS,
+    )
 
     scope = customer_memory_scope(context)
 
@@ -69,6 +106,7 @@ def test_scope_dedupes_and_skips_dirty_values() -> None:
         user_id="9002",
         roles=("risk_operator",),
         customer_ids=("9102", "9102", "not-a-number", "", "0", "-5", "9103"),
+        permissions=EMPLOYEE_PERMISSIONS,
     )
 
     # 脏值跳过而不是让整次召回崩掉；0 与负数不是合法客户号。

tests/unit/service/test_agent_governance.py

@@ -11,6 +11,7 @@ from app.core.contracts import (
     SourceReference,
 )
 from app.core.errors import ForbiddenAgentError, RecoverableAgentError
+from app.core.memory_scope import REQUIRED_EMPLOYEE_PERMISSION
 from app.service.agent import governance as governance_module
 from app.service.agent.base import BaseAgent
 from app.service.agent.customer_service_agent import CustomerServiceAgent
@@ -200,6 +201,14 @@ def _governance_with(
     return PlatformGovernance(recall_factory=lambda _session: service)  # type: ignore[arg-type,return-value]
 
 
+def _employee_context(user_id: str, customer_ids: tuple[str, ...]) -> RequestContext:
+    """员工身份上下文：带 `memory:read:customer` 能力码（读他人客户记忆的授权码）。"""
+    return RequestContext(
+        user_id=user_id, trace_id="t", roles=("advisor",), customer_ids=customer_ids,
+        permissions=(REQUIRED_EMPLOYEE_PERMISSION,),
+    )
+
+
 async def test_employee_recall_reads_assigned_customers(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -209,8 +218,7 @@ async def test_employee_recall_reads_assigned_customers(
         9102: [_StubItem("uuid-b", "持有约三年", 0.5)],
     })
     governance = _governance_with(monkeypatch, service)
-    context = RequestContext(user_id="9020", trace_id="t", roles=("advisor",),
-                             customer_ids=("9102", "9001"))
+    context = _employee_context("9020", ("9102", "9001"))
 
     memories = await governance.recall(context)
 
@@ -227,7 +235,7 @@ async def test_employee_without_assignment_reads_nothing_and_skips_database(
     """归属未维护 ⇒ 失败关闭，且**连召回服务都不该被调用**（不做无谓的库/向量查询）。"""
     service = _StubRecallService({9020: [_StubItem("uuid-self", "不该被读到", 0.9)]})
     governance = _governance_with(monkeypatch, service)
-    context = RequestContext(user_id="9020", trace_id="t", roles=("risk_operator",))
+    context = _employee_context("9020", ())
 
     memories = await governance.recall(context)
 
@@ -236,6 +244,23 @@ async def test_employee_without_assignment_reads_nothing_and_skips_database(
     assert service.queried == []
 
 
+async def test_employee_without_capability_reads_nothing(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """有归属行但缺 `memory:read:customer` 能力码 ⇒ 同样失败关闭，且不查库。"""
+    service = _StubRecallService({9001: [_StubItem("uuid-a", "不该被读到", 0.9)]})
+    governance = _governance_with(monkeypatch, service)
+    context = RequestContext(
+        user_id="9005", trace_id="t", roles=("operator",), customer_ids=("9001",),
+        permissions=("offsite:write",),
+    )
+
+    memories = await governance.recall(context)
+
+    assert memories == ()
+    assert service.queried == []
+
+
 async def test_customer_recall_ignores_assignments(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -257,8 +282,7 @@ async def test_customer_recall_ignores_assignments(
 def test_review_output_accepts_assigned_customer_memory_reference() -> None:
     """员工引用**归属客户**的记忆不算伪造引用；引用非归属客户的记忆必须被拦。"""
     config = ResolvedAgentConfig(config_version="1", prompt_version="p", model_endpoint="m")
-    context = RequestContext(user_id="9020", trace_id="t", roles=("advisor",),
-                             customer_ids=("9001",))
+    context = _employee_context("9020", ("9001",))
     recalled = (RecalledMemory(memory_uuid="uuid-a", customer_id="9001", content="偏好进取型"),)
 
     def _result(memory_uuid: str) -> AgentResult:
@@ -292,8 +316,6 @@ async def test_recall_scope_guard_rejects_memory_outside_assignments() -> None:
     agent.bind_governance(StubGovernance())  # type: ignore[arg-type]
     request = AgentRequest(agent_type="demo", message="看看客户", session_id="s",
                            idempotency_key="recall-scope-guard-0001")
-    context = RequestContext(user_id="9020", trace_id="t", roles=("advisor",),
-                             customer_ids=("9001",))
 
     with pytest.raises(RecoverableAgentError, match="越过客户范围"):
-        await agent.recall_memory(request, context)
+        await agent.recall_memory(request, _employee_context("9020", ("9001",)))