feat(pii_guard): PII打码守卫(2026-09-17用户定案合规)——手机/身份证MOD11-2/银行卡Luhn/邮箱确定性侦测→AES密文入Redis db4(24h ephemeral,不进机构密码表)→@@pii:占位符;gateway 0.6门禁+reply/progress用户侧还原+tool事件不还原;v2 run_command env注入$PIPELINE_PII_*+_save_turn落库掩码;人名NER留二期

pipeline_service/agent_loop_v2.py

@@ -139,10 +139,12 @@ class AgentExecutor:
         session_id: str = "",          # 会话内唯一标识（web 多 tab 独立会话；历史隔离键）
         default_pipeline_id: str = "", # 无当前项目时的默认产线（入口指定，如 bidding_general）
         delegate_depth: int = 0,       # 委派深度（0=顶层会话；子agent=1，禁止再委派）
+        pii_session_key: str = "",     # PII ephemeral 缓存键（gateway _session_key；空=无 PII 层）
     ):
         self.config = config
         self.project_id = project_id
         self.user_id = user_id
+        self.pii_session_key = pii_session_key
         self.org_id = ""              # loaded from project context
         self.pipeline_id = default_pipeline_id or ""  # 默认产线（入口指定）；有项目时被项目 pipeline_id 覆盖
         self.role = role               # 产线内角色（技能/工具/prompt 按角色加载）
@@ -1609,6 +1611,20 @@ class AgentExecutor:
             except Exception as e:
                 logger.warning(f"secret prepare_command failed (run without secrets): {e}")
 
+            # PII 执行边界（2026-09-17）：@@pii:占位符 → $PIPELINE_PII_*，真值经 env 注入，
+            # 与凭据同构；缓存缺失时引用保留（命令取空，不静默改语义——与凭据不同：PII
+            # 缺失是过期常态，拒绝执行会误伤，取空+输出可见更安全）。
+            pii_env = None
+            if self.pii_session_key and "@@pii:" in cmd:
+                try:
+                    from . import pii_guard
+                    cmd, pii_env = await pii_guard.prepare_command(
+                        cmd, self.pii_session_key)
+                except Exception as e:
+                    logger.warning(f"pii prepare_command failed (run without pii env): {e}")
+            if pii_env:
+                secret_env = {**(secret_env or {}), **pii_env}
+
             # 后台执行（2026-09-10，对齐 Hermes terminal background）：
             # 状态文件化到 workspace/.bg/，跨 worker 进程可 poll；
             # 沙箱档位与前台完全一致（generic 强制 strict，无 bwrap 上面已拒）。
@@ -2457,6 +2473,21 @@ class AgentExecutor:
         if self.config.session_isolation == "none":
             return
 
+        # PII 落库掩码（2026-09-17）：会话表是长期留存+session_search 可捞，占位符
+        # 虽非明文但缓存 24h 内可还原 → 落库一律换掩码形态（138****5678），
+        # 缓存过期后掩码仍是唯一可见形态（合规：留存数据不含可还原 PII）。
+        if self.pii_session_key and "@@pii:" in (user_input + reply):
+            try:
+                from . import pii_guard
+                if "@@pii:" in user_input:
+                    user_input = await pii_guard.restore_outbound(
+                        user_input, self.pii_session_key, masked=True)
+                if "@@pii:" in reply:
+                    reply = await pii_guard.restore_outbound(
+                        reply, self.pii_session_key, masked=True)
+            except Exception as e:
+                logger.warning(f"pii mask on save failed: {e}")
+
         try:
             from sqlor.dbpools import DBPools
             from appPublic.uniqueID import getID

pipeline_service/gateway.py

@@ -172,6 +172,22 @@ class Gateway:
         for note in secret_notes:
             yield json.dumps({"type": "progress", "message": note + "\n"}, ensure_ascii=False) + "\n"
 
+        # 0.6 PII 入站门禁（2026-09-17 用户定案：合规，敏感信息不传第三方）：
+        #     手机号/身份证/银行卡/邮箱等个人敏感信息 → AES 密文入 Redis（24h ephemeral，
+        #     **不进机构密码表**）→ 原文替换 @@pii:占位符。与 0.5 凭据门禁同构不同库。
+        #     放在模型/落库之前：净化后的 content 才是上行与落库版本。失败降级原文。
+        pii_key = self._session_key(channel, user_id, session_id)
+        try:
+            from . import pii_guard
+            _pr = await pii_guard.mask_inbound(content, pii_key)
+            content = _pr.get("text") or content
+            pii_notes = list(_pr.get("notes") or [])
+        except Exception as e:
+            logger.warning(f"pii inbound gate failed (continue with raw content): {e}")
+            pii_notes = []
+        for note in pii_notes:
+            yield json.dumps({"type": "progress", "message": note + "\n"}, ensure_ascii=False) + "\n"
+
         # 1. 解析项目上下文（纯通用模式跳过，不挂任何产线插件）
         if generic:
             ctx = {"pid": "", "pipeline_id": "", "name": "", "project_model_name": ""}
@@ -222,7 +238,7 @@ class Gateway:
         executor = AgentExecutor(
             config=config, project_id=ctx["pid"], user_id=user_id, role=role,
             base_url=base_url, generic=generic, session=sess, session_id=session_id,
-            default_pipeline_id=pipeline_id)
+            default_pipeline_id=pipeline_id, pii_session_key=pii_key)
 
         # 3.5 原生视觉（2026-09-10）：上传图片构造多模态 parts。
         #     上限/超限策略在 build_image_parts（8MB/张、4张/条，超限显式告知不静默丢）。
@@ -239,8 +255,21 @@ class Gateway:
             content = content + "\n\n[系统说明·本次上传图片] " + "；".join(image_notes)
 
         # 4. 转发标准化事件流
+        #    PII 用户侧还原（2026-09-17）：模型/落库只见占位符，回用户的 reply/progress
+        #    还原原文（用户自己贴的信息要能看见）；tool_call/tool_result **不还原**——
+        #    它们会进模型上下文与 llm_call_trace 上行。还原失败降级为占位符原样。
+        from . import pii_guard as _pii
         async for chunk in executor.run(content, history=history,
                                         image_parts=image_parts or None):
+            if pii_notes and chunk.strip():
+                try:
+                    _ev = json.loads(chunk.strip())
+                    if _ev.get("type") in ("reply", "progress") and "@@pii:" in str(_ev.get("message", "")):
+                        _ev["message"] = await _pii.restore_outbound(
+                            str(_ev["message"]), pii_key)
+                        chunk = json.dumps(_ev, ensure_ascii=False) + "\n"
+                except Exception:
+                    pass
             yield chunk
 
     async def _secret_inbound_gate(self, user_id: str, content: str):

pipeline_service/pii_guard.py

@@ -0,0 +1,258 @@
+"""pii_guard.py - 个人敏感信息（PII）打码守卫（2026-09-17 用户定案）。
+
+与凭据（secret_vault）**同构不同库**：
+  同构 —— 入站侦测 → 原文替换为占位符 `@@pii:KIND_FP@@` → 模型只见占位符 →
+          执行边界以子进程 env 注入真值（`$PIPELINE_PII_*`）→ 用户侧回复还原原文。
+  不同库 —— PII **不进机构密码表**（pipeline_user_secrets）：AES 密文存 Redis
+          （db4，TTL 24h，会话级 ephemeral），过期即焚，不做跨会话资产。
+
+语义差异（写死，不靠调用方自觉）：
+  凭据 = "只用不看"：还原只发生在执行 env，用户回复里也只见占位符/掩码。
+  PII  = "模型不看、用户看"：落库/上行/模型上下文全是占位符；回用户侧的
+         reply/progress 事件还原原文（用户自己贴的信息要能看见）。
+
+检测层只做**确定性格式**（手机号/身份证+MOD 11-2 校验/银行卡+Luhn/邮箱），
+不上模型：门禁在消息进模型之前，用 LLM 判断 = 先把原文发给 LLM，自相矛盾。
+人名/地址等无格式实体留二期（本地 NER，只标不杀，事后审计）。
+
+合规边界（一期）：覆盖 gateway v2 会话（入站门禁 + 用户侧还原 + run_command
+env 注入 + 会话落库占位符）。v1 角色 agent 交付件出库打码 = 二期，README 注明。
+"""
+import hashlib
+import json
+import re
+import time
+from typing import Dict, List, Optional, Tuple
+
+from appPublic.log import info, warning
+
+#: 占位符语法（与 @@sec: 同族不同前缀，互不干扰）
+PII_PLACEHOLDER_PREFIX = "@@pii:"
+PII_PLACEHOLDER_SUFFIX = "@@"
+PII_ENV_PREFIX = "PIPELINE_PII_"
+_PLACEHOLDER_RE = re.compile(r"@@pii:([A-Z]+_[0-9A-F]{8})@@")
+
+#: Redis 存储 TTL（24h，会话级 ephemeral；PII 不做长期资产）
+PII_TTL_SECONDS = 24 * 60 * 60
+_REDIS_DB = 4  # session 用 db3，PII 独立 db4，互不污染
+
+# ══════════════════════ 确定性检测器 ══════════════════════
+
+_ID_WEIGHTS = [7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2]
+_ID_CHECKCODES = "10X98765432"
+
+_RE_IDCARD = re.compile(r"(?<!\d)(\d{17}[\dXx])(?![\dXx])")
+_RE_PHONE = re.compile(r"(?<!\d)(1[3-9]\d{9})(?!\d)")
+_RE_BANKCARD = re.compile(r"(?<!\d)([3-6]\d{15,18})(?!\d)")
+_RE_EMAIL = re.compile(
+    r"([A-Za-z0-9._%+\-]{2,64}@[A-Za-z0-9.\-]+\.[A-Za-z]{2,24})")
+_EMAIL_JUNK_DOMAINS = ("example.com", "example.org", "test.com", "localhost",
+                       "domain.com", "email.com", "qq.com.fake")
+
+
+def idcard_checksum_ok(v: str) -> bool:
+    """GB 11643 MOD 11-2 校验。"""
+    v = v.upper()
+    if len(v) != 18:
+        return False
+    try:
+        s = sum(int(v[i]) * _ID_WEIGHTS[i] for i in range(17))
+    except ValueError:
+        return False
+    return _ID_CHECKCODES[s % 11] == v[17]
+
+
+def luhn_ok(v: str) -> bool:
+    digits = [int(c) for c in v]
+    odd = digits[-1::-2]
+    even = digits[-2::-2]
+    tot = sum(odd)
+    for d in even:
+        d *= 2
+        tot += d - 9 if d > 9 else d
+    return tot % 10 == 0
+
+
+def _mask(kind: str, v: str) -> str:
+    if kind == "PHONE":
+        return v[:3] + "****" + v[7:]
+    if kind == "IDCARD":
+        return v[:3] + "*" * (len(v) - 7) + v[-4:]
+    if kind == "BANKCARD":
+        return v[:4] + "*" * (len(v) - 8) + v[-4:]
+    if kind == "EMAIL":
+        local, _, domain = v.partition("@")
+        keep = local[:1] if local else ""
+        return keep + "***@" + domain
+    return "*" * len(v)
+
+
+def detect_pii(text: str) -> List[Dict]:
+    """扫描文本返回 PII 候选（span 去重，优先级 身份证>银行卡>手机>邮箱）。
+
+    返回项：{value, kind, masked, span}。只认确定性格式，零模型依赖。
+    """
+    if not text or not isinstance(text, str):
+        return []
+    hits: List[Dict] = []
+
+    def _add(kind: str, value: str, span: Tuple[int, int]):
+        hits.append({"kind": kind, "value": value,
+                     "masked": _mask(kind, value), "span": span})
+
+    for m in _RE_IDCARD.finditer(text):
+        v = m.group(1)
+        if idcard_checksum_ok(v):
+            _add("IDCARD", v.upper(), m.span(1))
+    for m in _RE_BANKCARD.finditer(text):
+        v = m.group(1)
+        if luhn_ok(v):
+            _add("BANKCARD", v, m.span(1))
+    for m in _RE_PHONE.finditer(text):
+        _add("PHONE", m.group(1), m.span(1))
+    for m in _RE_EMAIL.finditer(text):
+        v = m.group(1)
+        domain = v.partition("@")[2].lower()
+        if domain in _EMAIL_JUNK_DOMAINS or v.count("@") != 1:
+            continue
+        _add("EMAIL", v, m.span(1))
+
+    # span 去重：按优先级排序后剔除与已选区间重叠者（身份证 18 位优先于银行卡/手机子串）
+    prio = {"IDCARD": 0, "BANKCARD": 1, "PHONE": 2, "EMAIL": 3}
+    hits.sort(key=lambda h: (prio[h["kind"]], h["span"][0]))
+    chosen: List[Dict] = []
+    for h in hits:
+        a, b = h["span"]
+        if any(not (b <= ca or a >= cb) for ca, cb in
+               (c["span"] for c in chosen)):
+            continue
+        chosen.append(h)
+    chosen.sort(key=lambda h: h["span"][0])
+    return chosen
+
+
+def _ph_id(kind: str, value: str) -> str:
+    return kind + "_" + hashlib.sha1(value.encode("utf-8")).hexdigest()[:8].upper()
+
+
+def placeholder_of(kind: str, value: str) -> str:
+    return PII_PLACEHOLDER_PREFIX + _ph_id(kind, value) + PII_PLACEHOLDER_SUFFIX
+
+
+# ══════════════════════ Redis ephemeral 存储 ══════════════════════
+
+_redis_cli = None
+
+
+def _get_redis():
+    global _redis_cli
+    if _redis_cli is not None:
+        return _redis_cli
+    import redis.asyncio as aioredis
+    from appPublic.jsonConfig import getConfig
+    url = ""
+    try:
+        url = (getConfig().website.session_redis.url or "")
+    except Exception:
+        url = ""
+    if not url:
+        url = "redis://127.0.0.1:6379/3"
+    # 换 db 号：.../3 → .../4（PII 独立库）
+    base, _, db = url.rpartition("/")
+    url = (base + "/" + str(_REDIS_DB)) if base else url
+    _redis_cli = aioredis.Redis.from_url(url)
+    return _redis_cli
+
+
+def _rkey(session_key: str, ph_id: str) -> str:
+    return f"pii:{session_key}:{ph_id}"
+
+
+async def store_pii(session_key: str, kind: str, value: str, masked: str) -> str:
+    """AES 密文写 Redis（TTL 24h）。返回占位符。绝不写 DB 表。"""
+    from .secret_vault import encrypt_secret
+    r = _get_redis()
+    ph = placeholder_of(kind, value)
+    payload = json.dumps({"enc": encrypt_secret(value), "masked": masked,
+                          "kind": kind, "created": int(time.time())},
+                         ensure_ascii=False)
+    await r.set(_rkey(session_key, _ph_id(kind, value)), payload, ex=PII_TTL_SECONDS)
+    return ph
+
+
+async def _load(session_key: str, ph_id: str) -> Optional[Dict]:
+    r = _get_redis()
+    raw = await r.get(_rkey(session_key, ph_id))
+    if not raw:
+        return None
+    try:
+        if isinstance(raw, bytes):
+            raw = raw.decode("utf-8")
+        return json.loads(raw)
+    except Exception:
+        return None
+
+
+async def mask_inbound(text: str, session_key: str) -> Dict:
+    """入站门禁：侦测 → 密文入 Redis → 原文替换占位符。
+
+    返回 {text, notes, count}。任何异常由调用方兜底降级为原文。
+    """
+    cands = detect_pii(text)
+    if not cands:
+        return {"text": text, "notes": [], "count": 0}
+    out = text
+    kinds = []
+    for c in reversed(cands):  # 从后往前替换，span 不漂移
+        ph = await store_pii(session_key, c["kind"], c["value"], c["masked"])
+        out = out[:c["span"][0]] + ph + out[c["span"][1]:]
+        kinds.append(c["kind"])
+    note = ("检测到个人敏感信息（" + "、".join(sorted(set(kinds))) +
+            "）→ 已替换为占位符，原文加密缓存 24 小时（不进模型上下文、不明文落库），"
+            "回复中会自动还原给你。")
+    return {"text": out, "notes": [note], "count": len(cands)}
+
+
+async def restore_outbound(text: str, session_key: str, masked: bool = False) -> str:
+    """用户侧还原：占位符 → 原文（masked=False）或掩码形态（masked=True，日志/导出用）。
+
+    缓存过期/缺失 → 占位符原样保留（不猜不编，降级可见）。
+    """
+    if not text or PII_PLACEHOLDER_PREFIX not in text:
+        return text
+
+    async def _sub(m):
+        rec = await _load(session_key, m.group(1))
+        if not rec:
+            return m.group(0)
+        if masked:
+            return rec.get("masked", m.group(0))
+        from .secret_vault import decrypt_secret
+        return decrypt_secret(rec.get("enc", ""))
+
+    # 逐个替换（async re 不存在，手动遍历）
+    out = text
+    for m in reversed(list(_PLACEHOLDER_RE.finditer(text))):
+        repl = await _sub(m)
+        out = out[:m.start()] + repl + out[m.end():]
+    return out
+
+
+async def prepare_command(command: str, session_key: str) -> Tuple[str, Dict[str, str]]:
+    """执行边界：占位符 → `$PIPELINE_PII_*` env 引用 + env 真值字典。
+
+    与凭据 prepare_command 同构：命令字符串里**只有 env 引用**（落库/上行安全），
+    真值只经子进程 env 进入。缺失缓存 → 引用保留、env 不注入（命令自然取空）。
+    """
+    if not command or PII_PLACEHOLDER_PREFIX not in command:
+        return command, {}
+    env: Dict[str, str] = {}
+    out = command
+    for m in reversed(list(_PLACEHOLDER_RE.finditer(command))):
+        rec = await _load(session_key, m.group(1))
+        var = PII_ENV_PREFIX + m.group(1)
+        if rec:
+            from .secret_vault import decrypt_secret
+            env[var] = decrypt_secret(rec.get("enc", ""))
+            out = out[:m.start()] + "$" + var + out[m.end():]
+    return out, env