feat(chv): 对齐 Muse P6 privsep worker + 确定性安全检查层

- cloud/chv/privsep.py（新）：Gitea worker 注册表（凭证白名单只含
  GITEA_TOKEN）+ 类型化请求/响应（request_id 绑定、64KB 上限）+
  Unix socket 客户端（超时/崩溃 -> 类型化失败 WorkerRefusal）+
  客户端白名单校验
- cloud/chv/safety.py（新）：UntrustedData 来源标记 +
  InjectionScanner 确定性规则（block/flag/allow，中英）+
  redact_secrets + AuditLogger（只记指纹，不记值）+
  audit_inference_request/response
- chv-guest/chv-gitea-worker.py + .service（新）：systemd 沙箱化
  worker（SO_PEERCRED 只放 root，仓限 ~/workspace，git 只配 authd
  helper，绝不持 token；输出先扫描再返回）
- chv-guest/chv-sync-worker-client.py（新）：agent 侧薄 CLI
- chv-guest/chv-sync.sh：pull/push/status 有 socket 时走 worker，
  否则旧路径 + stderr 警告（过渡兼容）
- cloud/chv/executor.py：_run_guest 前后加推理边界审计钩子
  （best-effort）
- docs/chv-muse-p6-gap.md（新）：差距表 + 覆盖边界 + 真机待验证清单
- docs/chv-production-readiness.md：§15
- cloud/tests/test_chv_muse_p6.py（新）：43 项

测试：新测试 43/43；全量 cloud 套件 1910 OK（skipped=9）；mypy
在新增文件上干净（剩余 4 处为基线旧错 executor.py:1396/1401、
marker_tools.py:142，未动）。
真机验证：未做，排队在 round-11 之后，清单见
docs/chv-muse-p6-gap.md §4。

cloud/chv/safety.py

@@ -0,0 +1,281 @@
+"""确定性安全检查层（feat-chv-muse-p6，叶子模块）。
+
+对齐 Muse 官方安全文（security.muse.ai 2026-09-08，见
+docs/chv-muse-p6-gap.md）：Harness 层外部数据进模型上下文时标注为
+“不受信任输入”；独立分类器层多组提示注入检测分类器并行运行，覆盖
+经文件与工具调用进入上下文的全部外部数据；hatch-safety 检查进出
+核心模型推理的请求/响应。
+
+本模块是确定性规则版，不是模型分类器，不冒充（见差距表 §3.3）：
+同一输入永远同一 verdict，可复现、可测试；能拦什么、不能拦什么
+单列在差距表里。只用标准库，不依赖 cloud.*，客内 worker 守护进程
+可直接 import 它（见 chv-guest/chv-gitea-worker.py）。
+
+三件套：
+- 来源标记：`UntrustedData`（内容 + 来源 + 可信标志），随数据走；
+- 注入扫描：`InjectionScanner`，分级 block/flag/allow；
+- 推理边界审计：`AuditLogger`，只记指纹与 verdict，不记值；密钥
+  与 token 值先过 `redact_secrets` 再记，绝不进审计。
+"""
+
+import hashlib
+import logging
+import re
+import time
+from collections.abc import Callable
+
+_LOGGER = logging.getLogger("chv_safety")
+
+# 注入规则：(rule_id, tier, 正则)。tier 只分 block / flag 两档；
+# 无命中即 allow。block 只给高置信模式；flag 给中置信，命中即
+# 标记复核、不自动执行后续写操作。
+_RULES: tuple[tuple[str, str, str], ...] = (
+    # shell 执行链：curl/wget 的输出直接 piped 给 shell。
+    ("shell_pipe", "block",
+     r"(?i)\b(curl|wget)\b[^|\n]{0,120}\|\s*(sh|bash|zsh|dash)\b"),
+    # 命令替换里藏危险动词。
+    ("cmd_substitution", "block",
+     r"\$\([^)]*\b(rm|sudo|eval|exec|curl|wget|chmod|chown|kill|dd)\b"),
+    ("backtick_command", "block",
+     r"`[^`]*\b(rm|sudo|eval|exec|curl|wget)\b[^`]*`"),
+    # 直接指令覆盖（英文）。
+    ("instruction_override_en", "block",
+     r"(?i)\b(ignore|disregard|override|forget)\s+(all\s+)?"
+     r"(previous|prior|above|your|system)\s+(system\s+)?"
+     r"(instructions?|prompts?|rules?|guidelines?)"),
+    ("instruction_role_en", "block",
+     r"(?i)\byou are now\b.{0,40}\b(follow|obey|execute|do)\b"),
+    # 直接指令覆盖（中文）。
+    ("instruction_override_zh", "block",
+     r"忽略(之前|以前|上述)?(的)?(所有)?(的)?"
+     r"(指令|提示|指示|系统提示)"),
+    ("instruction_role_zh", "block",
+     r"你现在是.{0,20}(助手|模型|系统|AI|机器人)"),
+    # rm -rf 根目录 / 家目录（路径后不能再跟内容，/tmp/xxx 不算）。
+    ("rm_rf_root", "block",
+     r"\brm\s+-[a-z]*r[a-z]*f\s+(/|~|/home|/etc|/usr)(?![\w/])"),
+    # URL 里带 userinfo 口令（凭证泄露形态）。
+    ("url_userinfo", "block",
+     r"(?i)\bhttps?://[^\s/:@]+:[^\s@/]+@[^\s/]+"),
+    # data:/javascript: 伪协议。
+    ("dangerous_scheme", "block",
+     r"(?i)\b(data|javascript|vbscript):[^\s]*"),
+    # 代码执行原语（中置信：只 flag，不 block——正常文本里也可能
+    # 出现这些词，block 会误伤）。
+    ("code_exec_primitive", "flag",
+     r"(?i)\b(eval\s*\(|exec\s*\(|os\.system|subprocess\.|__import__\s*\()"),
+    # 指令性文本（中置信）。
+    ("instruction_like_en", "flag",
+     r"(?i)\b(follow these instructions|do not tell the user|"
+     r"keep this (secret|confidential)|do not (mention|reveal|disclose))"),
+    ("instruction_like_zh", "flag",
+     r"(不要告诉用户|不要让用户知道|秘密执行|不要透露|不要提及)"),
+    # 可疑 URL（中置信）：http 明文、IP 字面量主机、短链。
+    ("suspicious_url", "flag",
+     r"(?i)\bhttp://(?:\d{1,3}\.){3}\d{1,3}\b|"
+     r"\bhttps?://(?:bit\.ly|t\.co|tinyurl\.com|goo\.gl|is\.gd)/"),
+    # 可疑拼接：分隔线后紧跟指令性文本。
+    ("suspicious_concatenation", "flag",
+     r"(?m)^[-=*_]{3,}\s*$\n?.{0,200}?"
+     r"(ignore|disregard|you are now|忽略|你现在是)"),
+)
+
+# 脱敏模式：(正则, 替换)。只留键名/结构，值一律换成 [redacted]。
+_REDACT_RULES: tuple[tuple[str, str], ...] = (
+    (r"(?i)(gitea_token|token|password|passwd|secret|api[_-]?key|"
+     r"access[_-]?key)\s*[:=]\s*[\"']?\S+", r"\1=[redacted]"),
+    (r"sk-[A-Za-z0-9]{10,}", "[redacted]"),
+    (r"ghp_[A-Za-z0-9]{10,}", "[redacted]"),
+    (r"gho_[A-Za-z0-9]{10,}", "[redacted]"),
+    (r"github_pat_[A-Za-z0-9_]{10,}", "[redacted]"),
+    (r"(?i)Authorization:[^\n]*", "Authorization: [redacted]"),
+    (r"(?i)\bhttps?://[^\s/:@]+:[^\s@/]+@[^\s/]+",
+     "https://[redacted]@"),
+)
+
+_COMPILED_RULES: tuple[tuple[str, str, "re.Pattern[str]"], ...] = tuple(
+    (rule_id, tier, re.compile(pattern))
+    for rule_id, tier, pattern in _RULES
+)
+
+_COMPILED_REDACT: tuple[tuple["re.Pattern[str]", str], ...] = tuple(
+    (re.compile(pattern), repl) for pattern, repl in _REDACT_RULES
+)
+
+
+def redact_secrets(text: str) -> str:
+    """脱敏：把疑似密钥/口令的值换成 [redacted]，只留键名与结构。
+    输入非 str 即返回空串，绝不抛错。"""
+    if not isinstance(text, str):
+        return ""
+    out = text
+    for pattern, repl in _COMPILED_REDACT:
+        out = pattern.sub(repl, out)
+    return out
+
+
+def content_fingerprint(text: str) -> str:
+    """内容指纹（审计里唯一允许出现的内容派生形式）：先脱敏再
+    SHA-256，取前 16 位 hex。指纹不可逆推出原值。"""
+    cleaned = redact_secrets(text)
+    return hashlib.sha256(cleaned.encode("utf-8")).hexdigest()[:16]
+
+
+class Finding:
+    """一条规则命中：规则 id + 分级 + 截断后的证据片段（已脱敏）。"""
+
+    def __init__(self, rule: str, tier: str, excerpt: str) -> None:
+        self.rule = rule
+        self.tier = tier
+        self.excerpt = redact_secrets(excerpt)[:120]
+
+    def describe(self) -> dict:
+        """审计用描述（只含规则与分级，不含原文）。"""
+        return {"rule": self.rule, "tier": self.tier}
+
+
+class ScanResult:
+    """一次扫描结果：verdict + 命中列表。verdict 只取 block /
+    flag / allow 三档；block 命中一个即 block。"""
+
+    def __init__(self, verdict: str, findings: list) -> None:
+        self.verdict = verdict
+        self.findings = list(findings)
+
+    def describe(self) -> dict:
+        """审计用描述（只含 verdict 与规则名，不含原文）。"""
+        return {"verdict": self.verdict,
+                "rules": [f.rule for f in self.findings]}
+
+
+class InjectionScanner:
+    """确定性注入扫描器：规则全跑完再定级；同一输入永远同一
+    verdict。空文本即 allow，不抛错。"""
+
+    def scan(self, text: str) -> ScanResult:
+        """扫描一段外部文本，返回分级结果。"""
+        if not isinstance(text, str) or not text:
+            return ScanResult("allow", [])
+        findings = []
+        for rule_id, tier, pattern in _COMPILED_RULES:
+            match = pattern.search(text)
+            if match is not None:
+                findings.append(Finding(rule_id, tier, match.group(0)))
+        verdict = "allow"
+        if any(f.tier == "block" for f in findings):
+            verdict = "block"
+        elif findings:
+            verdict = "flag"
+        return ScanResult(verdict, findings)
+
+
+class UntrustedData:
+    """带来源标记的外部数据：内容 + 来源 + 可信标志 + 收到时间。
+
+    默认不受信任（trusted=False）；只有调用方核实过本地来源
+    （如本仓代码、本地配置）才用 mark_trusted。标记随数据走，
+    交给模型前须经 check_context_input。"""
+
+    def __init__(self, content: str, source: str, trusted: bool,
+                 received_at: float) -> None:
+        self.content = content
+        self.source = source
+        self.trusted = trusted
+        self.received_at = received_at
+
+    def describe(self) -> dict:
+        """审计用描述（只含来源与可信标志，不含内容）。"""
+        return {"source": self.source, "trusted": self.trusted,
+                "length": len(self.content),
+                "fingerprint": content_fingerprint(self.content)}
+
+
+def mark_untrusted(content: str, source: str) -> UntrustedData:
+    """把一段外部数据标成不受信任（来源如 gitea:<owner/repo>）。"""
+    return UntrustedData(content, source, False, time.time())
+
+
+def mark_trusted(content: str, source: str) -> UntrustedData:
+    """把一段已核实本地来源的数据标成可信。只给本地来源用；
+    远端拉回的数据一律走 mark_untrusted，绝不直接标可信。"""
+    return UntrustedData(content, source, True, time.time())
+
+
+def check_context_input(data: object,
+                        scanner: InjectionScanner | None = None
+                        ) -> ScanResult:
+    """数据进模型上下文前的检查入口：扫一遍再定级。
+
+    受信任的数据只 flag、不 block（本仓控制面自身的文本不应被
+    规则直接阻断，flag 进审计复核）；不受信任的数据按规则全分级。
+    返回的 ScanResult 由调用方执行：block 即阻断、flag 即标记复核。
+    """
+    text = ""
+    trusted = False
+    if isinstance(data, UntrustedData):
+        text = data.content
+        trusted = data.trusted
+    elif isinstance(data, str):
+        text = data
+    result = (scanner or InjectionScanner()).scan(text)
+    if trusted and result.verdict == "block":
+        capped = [Finding(f.rule, "flag", f.excerpt)
+                  for f in result.findings]
+        return ScanResult("flag", capped)
+    return result
+
+
+class AuditLogger:
+    """推理边界审计（hatch-safety 的确定性等价）：记进出核心模型
+    推理的请求/响应。只记任务 id、kind、内容指纹、长度、verdict；
+    内容值绝不进审计（先脱敏再算指纹）。emit 回调失败只记日志，
+    绝不抛错、不中断任务（沿用执行器 _emit 的既定语义）。"""
+
+    def __init__(self, emit: Callable[[dict], None] | None = None
+                 ) -> None:
+        self._emit = emit
+
+    def _record(self, event: str, task_id: str, kind: str,
+                content: str, verdict: str) -> dict:
+        record = {"event": event, "task_id": task_id, "kind": kind,
+                  "fingerprint": content_fingerprint(content),
+                  "length": len(content) if isinstance(content, str)
+                  else 0,
+                  "verdict": verdict}
+        try:
+            if self._emit is not None:
+                self._emit(record)
+            else:
+                _LOGGER.info("safety_audit %s", record)
+        except Exception as exc:  # 审计失败不中断任务
+            _LOGGER.error("safety_audit_failed event=%s error=%s",
+                          event, exc)
+        return record
+
+    def log_request(self, task_id: str, kind: str, content: str
+                    ) -> dict:
+        """记一次进推理的请求（任务 → 模型）。"""
+        return self._record("safety_inference_request", task_id,
+                            kind, content, "allow")
+
+    def log_response(self, task_id: str, kind: str, content: str,
+                     verdict: str = "allow") -> dict:
+        """记一次出推理的响应（模型 → 调用方），附扫描 verdict。"""
+        return self._record("safety_inference_response", task_id,
+                            kind, content, verdict)
+
+
+DEFAULT_AUDIT_LOGGER = AuditLogger()
+
+
+def audit_inference_request(task_id: str, kind: str, content: str
+                            ) -> dict:
+    """模块级快捷方式：记一次进推理的请求（默认 logger）。"""
+    return DEFAULT_AUDIT_LOGGER.log_request(task_id, kind, content)
+
+
+def audit_inference_response(task_id: str, kind: str, content: str,
+                             verdict: str = "allow") -> dict:
+    """模块级快捷方式：记一次出推理的响应（默认 logger）。"""
+    return DEFAULT_AUDIT_LOGGER.log_response(task_id, kind, content,
+                                             verdict)