Merge feat-chv-muse-p9: inference/telemetry restricted channel

- executor.py: P9 _inference_gate inserted before P1 dispatcher
  _run_with_vm; gate call is the first statement, so deny/ask
  fail-closed before any resident resume or cold wake. P9's own
  _run_with_vm (old-body variant) dropped: its body is byte-identical
  to P1 _run_cold (verbatim move), kept once in _run_cold.
- sentinel.py: P9 superset version wins (P5 base + inference entries
  POST /v1/responses, GET /v1/models + comments).
- inference_channel.py / telemetry.py / host.py: P9 as-is.
- Tests: P9 51 + closeup/gaps/optimize/cloud_hypervisor/fix 272 OK.
- mypy on changed files: P9's 5 cross-track errors gone (P5
  capabilities.py now in tree); remaining 4 are 06c3e17 baseline
  legacy (marker_tools.py:142, executor.py:1573/1578), untouched.

cloud/chv/executor.py

@@ -2727,6 +2727,54 @@ class CloudHypervisorExecutor(Executor):
             except Exception:
                 pass
 
+    def _inference_gate(self, task: dict,
+                        model_provider: str | None,
+                        extra_env: dict | None = None) -> None:
+        """feat-chv-muse-p9：起 VM 前的推理出口预审。
+
+        模型调用只允许走受限推理通道（见 inference_channel）：
+        allow → 审计放行；deny/ask → 审计 + ChvFailure 起 VMM
+        前 fail-closed（ask 在这里就是 deny：wake 等不了审批
+        UI）。解析器可注入（`sentinel_resolver` 属性，测试用
+        假解析器）。origin 取任务级 extra_env 覆盖执行器 env；
+        origin 非法同样记 deny，不抛裸 ValueError。审计 detail
+        只含 VM 名/主机/verdict，不含任何凭证值。
+        """
+        from cloud.chv.inference_channel import check_inference_call
+        name = self.vm_name(task)
+        env = dict(self._env)
+        if extra_env:
+            env.update(extra_env)
+        try:
+            check = check_inference_call(
+                model_provider, env=env,
+                resolve=getattr(self, "sentinel_resolver", None))
+        except ValueError as exc:
+            self._emit("inference_egress_deny", task,
+                       "vm=%s reasons=bad_origin" % name)
+            raise ChvFailure(
+                "inference_egress_denied",
+                "推理 origin 配置非法，拒绝启动：%s" % exc)
+        decision = check.decision
+        if decision.verdict == "allow":
+            capability = decision.capability
+            request = check.request
+            assert capability is not None  # allow 恒带严格能力
+            assert request is not None  # allow 恒带已解析请求
+            self._emit("inference_egress_allow", task,
+                       "vm=%s host=%s capability=%s"
+                       % (name, request.host, capability.id))
+            return
+        host = check.request.host if check.request is not None else ""
+        self._emit("inference_egress_" + decision.verdict, task,
+                   "vm=%s host=%s reasons=%s"
+                   % (name, host, ",".join(decision.reasons)))
+        raise ChvFailure(
+            "inference_egress_denied",
+            "推理出口预审未通过（%s），拒绝启动：请运维确认 "
+            "provider 与 CHV_INFERENCE_ORIGIN/CHV_INFERENCE_ALLOW_HOSTS"
+            % decision.verdict)
+
     def _run_with_vm(self, task, inner_args, goal=False, extra_env=None,
                      **kwargs):
         """任务 VM 生命周期（feat-chv-muse-p1，见 ADR-17）：
@@ -2740,7 +2788,11 @@ class CloudHypervisorExecutor(Executor):
         重试（重试是 worker 的 max_retries 语义）。中途死亡
         判 interrupted（断点可续跑），见 _run_cold 与
         _run_on_resident。dispatch/run/goal/goal_resume/断点
-        续跑都走本方法，签名不动。"""
+        续跑都走本方法，签名不动。round-12 合入 P9：起任何 VM
+        （常驻 resume 或冷启动 wake）之前先过 _inference_gate
+        推理出口预审，deny/ask 即 fail-closed。"""
+        self._inference_gate(task, kwargs.get("model_provider"),
+                             extra_env=extra_env)
         resident = self._resident_acquire(task, extra_env=extra_env)
         if resident is not None:
             pool = resident["pool"]

cloud/chv/inference_channel.py

@@ -0,0 +1,247 @@
+"""受限推理通道（feat-chv-muse-p9，叶子模块）。
+
+对齐 Muse 官方安全文（security.muse.ai 2026-09-08，见
+docs/chv-muse-p9-gap.md）：推理与遥测代理是通向外部基础设施的
+受限通道；Sentinel 是连接器动作与全部网络出口的唯一权限机构。
+
+本模块是**决策层**，不是网络层强制：
+- 它不拦截客内流量。客内 codex 直连的强制仍靠 Tinyproxy 白名单 +
+  宿主防火墙（第三棒落地中）；不要把本模块写成“出口已全部经
+  Sentinel”——与 P3 的 sentinel 模块同款声明。
+- 唯一出口点：模型调用只允许走这里决定的 origin。执行器在
+  `_run_with_vm` 起 VM 前调 `check_inference_call`：allow 才继续；
+  deny/ask 一律审计 + 停止，不起 VM。
+- provider=`sub2api` 时 origin 取 `SUB2API_BASE_URL`；provider 为空
+  （xcmapi 路径，codex 用自身 config.toml 默认网关）时执行器看不
+  到 origin，改取 operator 显式配置的 `CHV_INFERENCE_ORIGIN`，未配
+  即 ask（fail-closed，不静默放行）。部署迁移见差距表 §2.5。
+- 能力说明：allow 时带出的严格能力（单次、300 秒）是**起 VM 前的
+  预审授权**，不是逐次 HTTP 请求的令牌；客内 codex 的逐次调用仍由
+  代理白名单 + 防火墙约束。本模块不伪装成逐请求强制。
+
+本模块只用标准库，不依赖 cloud.*（除 vendored 的 sentinel），
+解析器可注入（`resolve` 参数），默认 `socket.getaddrinfo`；测试
+用假解析器，绝不联网。审计记录里绝不含凭证值。
+"""
+
+import ipaddress
+import os
+from collections.abc import Callable
+from typing import NamedTuple
+from urllib.parse import urlparse
+
+from cloud.chv.sentinel import (
+    Decision,
+    EgressRequest,
+    SentinelPolicy,
+    _one_shot_capability,
+)
+
+INFERENCE_CONNECTOR = "inference"
+
+# 推理连接器的 L7 路径规则：codex 以 wire_api="responses" 调模型，
+# 只放 POST /v1/responses；GET /v1/models 是否必需待真机核实
+# （codex 启动时是否拉模型列表），若非必需则收紧为只放前者。
+INFERENCE_PATH_RULES = (
+    ("POST", "prefix", "/v1/responses"),
+    ("GET", "prefix", "/v1/models"),
+)
+
+# 上限默认值（本项目自定，非 Meta 官方值）：超时 120 秒与
+# host.py 的 Server.request 默认超时一致；重试 1 次即保持
+# 现有行为不变（现为 range(2)）。非法值一律 ValueError。
+DEFAULT_INFERENCE_TIMEOUT_SECONDS = 120
+DEFAULT_INFERENCE_MAX_RETRIES = 1
+MIN_INFERENCE_TIMEOUT_SECONDS = 10
+MAX_INFERENCE_TIMEOUT_SECONDS = 600
+MAX_INFERENCE_RETRIES = 5
+
+
+class InferenceLimits(NamedTuple):
+    """推理通道上限：单次请求超时（秒）与重连重试上限（次）。"""
+
+    timeout_seconds: int
+    max_retries: int
+
+
+class InferenceCheck(NamedTuple):
+    """一次预审的结果：决策 + 被评估的请求（origin 缺失/非法时
+    request 为 None，审计方用原因兜底）。"""
+
+    decision: Decision
+    request: EgressRequest | None
+
+
+def _env_of(env: dict | None) -> dict:
+    """测试传 dict，生产默认读 os.environ。"""
+    return env if env is not None else os.environ  # type: ignore[return-value]
+
+
+def _is_local_relay_host(host: str) -> bool:
+    """字面环回/私网 IP 才是本地中继主机（无 DNS，无重绑定；
+    主机名一律不是——DNS 解析结果不可信）。"""
+    try:
+        addr = ipaddress.ip_address(host)
+    except ValueError:
+        return False
+    return addr.is_loopback or addr.is_private
+
+
+def resolve_inference_origin(provider: str | None,
+                             env: dict | None = None) -> str:
+    """解析推理出口 origin（不带尾斜杠，端口非默认时保留）。
+
+    - provider="sub2api"：取 `SUB2API_BASE_URL`；
+    - provider 为空或未知标签：取 `CHV_INFERENCE_ORIGIN`（与
+      host.py `_server_command` 一致走默认网关；标签本身不决定
+      origin，origin 永远只取 operator 配置的 env）；
+    变量未配/空白返回 ""（调用方按 ask 处理，不在这里抛错）。
+    scheme 规则（见 cloud/README.md 的 sub2api 网关一节）：
+    - https：公网网关（如 xgapi.lol），端口只接受 443（或缺省）；
+    - http：只接受本地中继——字面环回/私网 IP 主机（如文档形状
+      http://172.17.0.1:28084，Docker 桥接 IP），端口保留；
+    - 其余（http 公网主机名、https 非 443 端口、带 userinfo、
+      无 host）一律 ValueError——配置错了要 loud，不要静默跑向别处。
+    报错不带 origin 原值（http URL 里可能含 userinfo）。
+    """
+    values = _env_of(env)
+    if provider is None:
+        raw = values.get("CHV_INFERENCE_ORIGIN") or ""
+    elif provider == "sub2api":
+        raw = values.get("SUB2API_BASE_URL") or ""
+    else:
+        # 未知 provider 标签：与 host.py `_server_command` 一致走
+        # 默认网关（`CHV_INFERENCE_ORIGIN`）——标签本身不决定
+        # origin，origin 永远只取 operator 配置的 env。
+        raw = values.get("CHV_INFERENCE_ORIGIN") or ""
+    raw = raw.strip().rstrip("/")
+    if not raw:
+        return ""
+    parsed = urlparse(raw)
+    scheme = parsed.scheme.lower()
+    if "@" in parsed.netloc:
+        raise ValueError("推理 origin 不得带 userinfo（凭证禁进 URL）")
+    host = (parsed.hostname or "").lower()
+    if not host:
+        raise ValueError("推理 origin 无主机名")
+    if scheme == "https":
+        if parsed.port is not None and parsed.port != 443:
+            raise ValueError("推理 origin 端口只接受 443")
+        return "https://" + host
+    if scheme == "http":
+        if not _is_local_relay_host(host):
+            raise ValueError("http 推理 origin 只接受本地中继"
+                             "（环回/私网 IP）")
+        port = parsed.port
+        if port is not None:
+            return "http://%s:%d" % (host, port)
+        return "http://" + host
+    raise ValueError("推理 origin 必须 https（或本地中继 http）")
+
+
+def inference_allow_hosts(provider: str | None,
+                          env: dict | None = None) -> frozenset:
+    """推理出口 allowlist（小写主机名集合）。
+
+    `CHV_INFERENCE_ALLOW_HOSTS`（逗号分隔）显式覆盖；未配时默认
+    即 origin 自身的主机——allowlist 永远是“显式配置的 origin”，
+    不是“解析后是什么都行”。origin 缺失时返回空集（预审必 ask）。
+    """
+    values = _env_of(env)
+    raw = values.get("CHV_INFERENCE_ALLOW_HOSTS") or ""
+    hosts = frozenset(h.strip().lower()
+                      for h in raw.split(",") if h.strip())
+    if hosts:
+        return hosts
+    try:
+        origin = resolve_inference_origin(provider, env)
+    except ValueError:
+        return frozenset()
+    if not origin:
+        return frozenset()
+    host = urlparse(origin).hostname or ""
+    return frozenset({host}) if host else frozenset()
+
+
+def _parse_int_env(env: dict, name: str, default: int,
+                   low: int, high: int) -> int:
+    """严格整数 env 解析：未配即默认；非整数/越界即 ValueError，
+    绝不静默回落（配错早炸，不留到起 VM 才暴露）。"""
+    raw = env.get(name)
+    if raw is None or str(raw).strip() == "":
+        return default
+    try:
+        value = int(str(raw).strip())
+    except ValueError:
+        raise ValueError("%s 必须是整数: %r" % (name, raw))
+    if not low <= value <= high:
+        raise ValueError("%s 必须在 %d-%d 之间: %s"
+                         % (name, low, high, value))
+    return value
+
+
+def inference_limits(env: dict | None = None) -> InferenceLimits:
+    """推理通道上限（见模块头默认值说明）。"""
+    values = _env_of(env)
+    return InferenceLimits(
+        timeout_seconds=_parse_int_env(
+            values, "CHV_INFERENCE_TIMEOUT",
+            DEFAULT_INFERENCE_TIMEOUT_SECONDS,
+            MIN_INFERENCE_TIMEOUT_SECONDS,
+            MAX_INFERENCE_TIMEOUT_SECONDS),
+        max_retries=_parse_int_env(
+            values, "CHV_INFERENCE_MAX_RETRIES",
+            DEFAULT_INFERENCE_MAX_RETRIES, 0, MAX_INFERENCE_RETRIES),
+    )
+
+
+def check_inference_call(provider: str | None,
+                         env: dict | None = None,
+                         resolve: Callable[[str, int], list] | None = None,
+                         policy: SentinelPolicy | None = None,
+                         now: float | None = None) -> InferenceCheck:
+    """预审一次推理出口调用（起 VM 前，fail-closed）。
+
+    - origin 缺失 → ask（`inference_origin_not_configured`），
+      调用方停止执行，不得自动放行；
+    - origin 非法 → deny；
+    - http 本地中继（字面环回/私网 IP）：无 DNS、无重绑定，跳过
+      Sentinel 的 DNS/SSRF 步；allowlist 与单次能力照常执行；
+    - 其余走 Sentinel 评估：SSRF、私网 IP、scheme/端口/路径越界
+      即 deny；主机不在白名单即 ask；全过即 allow + 严格能力
+      （单次、300 秒，见模块头的能力说明）。
+    `resolve` / `policy` 可注入（假解析器 / P5 纵深策略），默认按 allowlist 现建。
+    """
+    try:
+        origin = resolve_inference_origin(provider, env)
+    except ValueError:
+        # 原因不带原值：非法 origin 的原串里可能含 userinfo，
+        # 审计原因只记“非法”本身，不记值。
+        return InferenceCheck(
+            Decision("deny", ["inference_origin_invalid"]), None)
+    if not origin:
+        return InferenceCheck(
+            Decision("ask", ["inference_origin_not_configured"]), None)
+    parsed = urlparse(origin)
+    scheme = parsed.scheme
+    host = parsed.hostname or ""
+    port = parsed.port or (443 if scheme == "https" else 80)
+    if policy is None:
+        policy = SentinelPolicy(
+            allow_hosts=inference_allow_hosts(provider, env),
+            connector_paths={INFERENCE_CONNECTOR: INFERENCE_PATH_RULES})
+    request = EgressRequest(INFERENCE_CONNECTOR, "POST", scheme,
+                            host, port, "/v1/responses")
+    if scheme == "http":
+        # 本地中继分支：见上。allowlist 不含该主机 → ask；
+        # 含 → allow + 与公网路径同款单次能力（P3 默认档）。
+        if host not in policy.allow_hosts:
+            return InferenceCheck(
+                Decision("ask", ["host_not_allowlisted"]), request)
+        rules = policy.connector_paths.get(INFERENCE_CONNECTOR, ())
+        capability = _one_shot_capability(request, rules, now)
+        return InferenceCheck(
+            Decision("allow", ["policy_allow", "local_relay"],
+                     capability=capability), request)
+    decision = policy.evaluate(request, resolve=resolve, now=now)
+    return InferenceCheck(decision, request)

cloud/chv/sentinel.py

@@ -95,6 +95,10 @@ _BLOCKED_NETWORKS = tuple(
 # gitea：API 只放 /api/v1/；git 智能 HTTP 只放三条（clone/pull/push
 # 的实际路径，见 git http-protocol）。其它路径（Web UI、登录页、
 # 任意管理接口）一律不在本连接器的能力内。
+# feat-chv-muse-p9 的 inference 条目：推理连接器只放
+# POST /v1/responses 与 GET /v1/models（/v1/models 是否必需待
+# 真机核实）；scheme/端口见下面两表。本文件其余内容与 P5 的
+# 25bcd35 一致，合入时以本文件为准（它是 P3+P5+P9 的并集）。
 _CONNECTOR_PATH_RULES = {
     "gitea": (
         ("GET", "prefix", "/api/v1/"),
@@ -103,12 +107,21 @@ _CONNECTOR_PATH_RULES = {
         ("POST", "suffix", ".git/git-upload-pack"),
         ("POST", "suffix", ".git/git-receive-pack"),
     ),
+    "inference": (
+        ("POST", "prefix", "/v1/responses"),
+        ("GET", "prefix", "/v1/models"),
+    ),
 }
 
 # 每连接器允许的 scheme/端口（默认：只 https/443；http 只在
-# 显式放行时才出现，本期没有任何连接器放 http）。
+# 显式放行时才出现，本期没有任何连接器放 http）。注意：推理的
-_CONNECTOR_SCHEMES = {"gitea": frozenset({"https"})}
+# http 本地中继（字面环回/私网 IP，如 http://172.17.0.1:28084）
-_CONNECTOR_PORTS = {"gitea": frozenset({443})}
+# 不走 evaluate（见 inference_channel 的本地中继分支），不要
+# “顺手”把 http 加进下面两表——那会放宽公网路径的 scheme 检查。
+_CONNECTOR_SCHEMES = {"gitea": frozenset({"https"}),
+                      "inference": frozenset({"https"})}
+_CONNECTOR_PORTS = {"gitea": frozenset({443}),
+                    "inference": frozenset({443})}
 
 # 默认放凭证有效期（秒）：与 authd 按次释放同款 300 秒。
 DEFAULT_CAPABILITY_TTL_SECONDS = 300

cloud/chv/telemetry.py

@@ -0,0 +1,211 @@
+"""遥测最小化 + 训练退出开关（feat-chv-muse-p9，叶子模块）。
+
+对齐 Muse 官方安全文（security.muse.ai 2026-09-08，见
+docs/chv-muse-p9-gap.md）：VM 发送有限数据用于推理与遥测
+（安全文未给字段级清单——本模块的字段清单按最小化原则自定，
+见 `TELEMETRY_FIELDS`）；推理轨迹在去除关键个人身份信息后可
+用于训练新检查点，用户可一键退出训练。
+
+诚实边界（见差距表 §4）：
+- 遥测目前只落盘 `<state_dir>/telemetry.jsonl`，没有向外部遥测
+  端点发送的路径；落盘失败也不影响任务（`emit_telemetry`
+  永不抛错）。
+- 训练轨迹目前没有上传路径：开关管的是“轨迹类事件是否产生”；
+  上传通道本身不存在，开关是为上传通道预留的门禁。
+- 字段白名单是构造性保证：不在白名单里的键一律丢弃；token 数
+  只取 host.py 回传的 usage，未回传记 `null`，不估算、不编造。
+- 训练开关默认开是**本项目决定，非 Meta 官方值**（报告注明
+  帮助中心原文未逐字确认默认开启）。
+
+本模块只用标准库 + `atomic_file`（开关状态文件原子写）。
+"""
+
+import hashlib
+import json
+import logging
+import os
+import time
+
+from cloud.chv.atomic_file import atomic_write_text
+
+_LOGGER = logging.getLogger(__name__)
+
+# 遥测字段白名单（自定，按最小化原则）：只含任务级元数据。
+# 绝不含对话原文、文件内容、密钥、个人身份信息——白名单外
+# 的键在 sanitize 时一律丢弃，不是“记下来再脱敏”。
+TELEMETRY_FIELDS = frozenset({
+    "task_id", "vm_id", "provider", "model", "status",
+    "started_at", "ended_at", "duration_ms",
+    "tokens_in", "tokens_out", "tokens_total",
+    "turns", "tool_calls", "sampled",
+})
+
+# 轨迹类事件字段白名单（自定）：只含回合级元数据，不含任何文本。
+TRAJECTORY_FIELDS = frozenset({
+    "task_id", "turns", "tools", "duration_ms",
+})
+
+TRAINING_OPT_FILENAME = "training-opt.json"
+TRAINING_OPT_AUDIT_FILENAME = "training-opt-audit.jsonl"
+TELEMETRY_FILENAME = "telemetry.jsonl"
+DEFAULT_TELEMETRY_SAMPLE_RATE = 1.0
+
+
+def telemetry_sample_rate(env: dict | None = None) -> float:
+    """遥测采样率：`CHV_TELEMETRY_SAMPLE_RATE`，0.0–1.0，默认 1.0
+    （全量，任务级元数据量小）。非法值即 ValueError。"""
+    values = env if env is not None else os.environ  # type: ignore[assignment]
+    raw = values.get("CHV_TELEMETRY_SAMPLE_RATE")
+    if raw is None or str(raw).strip() == "":
+        return DEFAULT_TELEMETRY_SAMPLE_RATE
+    try:
+        rate = float(str(raw).strip())
+    except ValueError:
+        raise ValueError(
+            "CHV_TELEMETRY_SAMPLE_RATE 必须是数字: %r" % raw)
+    if not 0.0 <= rate <= 1.0:
+        raise ValueError(
+            "CHV_TELEMETRY_SAMPLE_RATE 必须在 0.0-1.0 之间: %s" % rate)
+    return rate
+
+
+def sampled(task_id: str, rate: float) -> bool:
+    """确定性采样：同一 task_id 每次判定一致（可测试）；rate=0
+    全丢，rate=1 全留。"""
+    digest = hashlib.sha256(task_id.encode("utf-8")).hexdigest()
+    return int(digest[:8], 16) % 10000 < rate * 10000
+
+
+def _is_scalar(value: object) -> bool:
+    """遥测值只认标量（str/int/float/bool/None）；dict/list
+    一律丢弃——不把结构化内容带进遥测。"""
+    return value is None or isinstance(value, (str, int, float, bool))
+
+
+def sanitize_telemetry_event(event: dict) -> dict:
+    """按白名单过滤遥测事件：不在白名单里的键丢弃；值不是标量
+    的丢弃；`tools` 字段例外——只认 str 列表（工具名），其它
+    元素一律去掉。返回新 dict，不改原 event。"""
+    clean: dict = {}
+    for key, value in (event or {}).items():
+        if key not in TELEMETRY_FIELDS and key not in TRAJECTORY_FIELDS:
+            continue
+        if key == "tools":
+            if isinstance(value, (list, tuple)):
+                clean[key] = [t for t in value if isinstance(t, str)]
+            continue
+        if _is_scalar(value):
+            clean[key] = value
+    return clean
+
+
+def build_task_telemetry(task_id: str, vm_id: str, provider: str,
+                         model: str | None, status: str,
+                         started_at: float, ended_at: float,
+                         duration_ms: int,
+                         tokens: dict | None = None,
+                         turns: int | None = None,
+                         tool_calls: int | None = None,
+                         env: dict | None = None) -> dict | None:
+    """建一条任务级遥测事件；采样未命中返回 None（不产生事件，
+    不是产生后丢弃）。tokens 取 host.py 回传的 usage，未回传
+    时三个 token 字段记 None，不编造。"""
+    rate = telemetry_sample_rate(env)
+    if not sampled(task_id, rate):
+        return None
+    tokens = tokens or {}
+    event = {
+        "task_id": task_id, "vm_id": vm_id, "provider": provider,
+        "model": model, "status": status,
+        "started_at": started_at, "ended_at": ended_at,
+        "duration_ms": duration_ms,
+        "tokens_in": tokens.get("input_tokens"),
+        "tokens_out": tokens.get("output_tokens"),
+        "tokens_total": tokens.get("total_tokens"),
+        "turns": turns, "tool_calls": tool_calls, "sampled": True,
+    }
+    return sanitize_telemetry_event(event)
+
+
+def build_trajectory_event(task_id: str, turns: int,
+                           tools: list, duration_ms: int,
+                           state_dir: str) -> dict | None:
+    """建一条轨迹类事件（训练用）：只含回合级元数据（回合数、
+    工具名、耗时），不含任何文本。训练退出开关关时返回 None
+    （事件不再产生）。"""
+    if not training_enabled(state_dir):
+        return None
+    return sanitize_telemetry_event({
+        "task_id": task_id, "turns": turns,
+        "tools": list(tools or []), "duration_ms": duration_ms,
+    })
+
+
+def _append_jsonl(path: str, record: dict) -> bool:
+    """best-effort 追加一行 JSONL：失败返回 False，永不抛错
+    （遥测/审计落盘失败绝不影响任务）。"""
+    try:
+        line = json.dumps(record, ensure_ascii=False) + "\n"
+        with open(path, "a", encoding="utf-8") as fh:
+            fh.write(line)
+            fh.flush()
+        return True
+    except Exception as exc:  # noqa: BLE001 - 落盘失败只记日志
+        _LOGGER.warning("telemetry_append_failed path=%s error=%s",
+                        path, exc)
+        return False
+
+
+def emit_telemetry(state_dir: str, event: dict) -> bool:
+    """发一条遥测事件：先过白名单再落盘 `<state_dir>/telemetry.jsonl`。
+    落盘失败返回 False，永不抛错、不影响任务。"""
+    if not state_dir or not isinstance(event, dict):
+        return False
+    clean = sanitize_telemetry_event(event)
+    if not clean:
+        return False
+    return _append_jsonl(
+        os.path.join(state_dir, TELEMETRY_FILENAME), clean)
+
+
+def training_enabled(state_dir: str) -> bool:
+    """训练退出开关状态：默认开（本项目决定，非 Meta 官方值）。
+    状态文件缺失即默认开；损坏/不可解析/缺 `enabled` 键时
+    fail-closed 当作关处理并记日志（损坏不等于同意上传）。"""
+    path = os.path.join(state_dir, TRAINING_OPT_FILENAME)
+    try:
+        with open(path, encoding="utf-8") as fh:
+            data = json.load(fh)
+    except FileNotFoundError:
+        return True
+    except Exception as exc:  # noqa: BLE001 - 损坏即关
+        _LOGGER.warning("training_opt_corrupt path=%s error=%s",
+                        path, exc)
+        return False
+    if not isinstance(data, dict) or "enabled" not in data:
+        return False
+    return bool(data["enabled"])
+
+
+def set_training_enabled(state_dir: str, enabled: bool,
+                         actor: str = "user") -> bool:
+    """一键退出/恢复训练：状态原子写 `<state_dir>/training-opt.json`
+    （0600），变更追加进 `training-opt-audit.jsonl`。`enabled`
+    不是 bool 即 TypeError；写失败返回 False，永不抛错。
+    关后立即生效：后续轨迹事件不再产生。"""
+    if not isinstance(enabled, bool):
+        raise TypeError("enabled 必须是 bool: %r" % (enabled,))
+    record = {"enabled": enabled, "updated_at": time.time(),
+              "actor": str(actor)[:64]}
+    try:
+        atomic_write_text(
+            os.path.join(state_dir, TRAINING_OPT_FILENAME),
+            json.dumps(record, ensure_ascii=False))
+    except Exception as exc:  # noqa: BLE001 - 写失败只记日志
+        _LOGGER.warning("training_opt_write_failed error=%s", exc)
+        return False
+    audit = dict(record)
+    audit["event"] = "training_opt_changed"
+    _append_jsonl(
+        os.path.join(state_dir, TRAINING_OPT_AUDIT_FILENAME), audit)
+    return True

cloud/tests/test_chv_closeup.py

@@ -85,6 +85,12 @@ def _write(path, content=b"x"):
     return path
 
 
+def _public_resolver(host, port):
+    """feat-chv-muse-p9：假解析器（公网 IP），供推理出口预审用；
+    实验室绝不联网，真实解析只在真机发生。"""
+    return ["93.184.216.34"]
+
+
 class CloseupCase(unittest.TestCase):
     def setUp(self):
         self.tmp = tempfile.TemporaryDirectory()
@@ -113,7 +119,11 @@ class CloseupCase(unittest.TestCase):
                   env_extra=None, **overrides):
         env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888",
                "CHV_HOST_KEY_MODE": "keyscan",
-               "CHV_SSH_FORWARD": "1"}
+               "CHV_SSH_FORWARD": "1",
+               # feat-chv-muse-p9：推理出口预审的假 origin（公网
+               # https，配假解析器；实验室不起真 VM、不联网）。
+               "SUB2API_BASE_URL": "https://xgapi.lol",
+               "CHV_INFERENCE_ORIGIN": "https://xgapi.lol"}
         if env_extra:
             env.update(env_extra)
         kwargs = dict(
@@ -127,6 +137,7 @@ class CloseupCase(unittest.TestCase):
             sleeper=lambda s: None, env=env)
         kwargs.update(overrides)
         ex = CloudHypervisorExecutor(**kwargs)
+        ex.sentinel_resolver = _public_resolver
         ex.audit_callback = lambda e, t, d, k=None: self.audit.append(
             (e, t.get("id") if isinstance(t, dict) else t, d, k))
         return ex

cloud/tests/test_chv_fix.py

@@ -73,6 +73,12 @@ def _write(path, content=b"x"):
     return path
 
 
+def _public_resolver(host, port):
+    """feat-chv-muse-p9：假解析器（公网 IP），供推理出口预审用；
+    实验室绝不联网，真实解析只在真机发生。"""
+    return ["93.184.216.34"]
+
+
 class ChvFixCase(unittest.TestCase):
     def setUp(self):
         self.tmp = tempfile.TemporaryDirectory()
@@ -127,7 +133,11 @@ class ChvFixCase(unittest.TestCase):
 
     def make_exec(self, runner=None, env_extra=None, **overrides):
         env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888",
-               "CHV_SSH_FORWARD": "1"}
+               "CHV_SSH_FORWARD": "1",
+               # feat-chv-muse-p9：推理出口预审的假 origin（公网
+               # https，配假解析器；实验室不起真 VM、不联网）。
+               "SUB2API_BASE_URL": "https://xgapi.lol",
+               "CHV_INFERENCE_ORIGIN": "https://xgapi.lol"}
         if env_extra:
             env.update(env_extra)
         kwargs = dict(
@@ -141,6 +151,7 @@ class ChvFixCase(unittest.TestCase):
             sleeper=lambda s: None, env=env)
         kwargs.update(overrides)
         ex = CloudHypervisorExecutor(**kwargs)
+        ex.sentinel_resolver = _public_resolver
         ex.audit_callback = lambda e, t, d, k=None: self.audit.append(
             (e, t.get("id") if isinstance(t, dict) else t, d, k))
         return ex
@@ -305,8 +316,12 @@ class GatewayExtraEnvTest(ChvFixCase):
         task = {"id": "gw1", "project_id": "p1", "prompt": "做目标",
                 "goal_mode": 1,
                 "_gateway_provider": "sub2api",
+                # feat-chv-muse-p9：网关 URL 须是合法推理 origin——
+                # 这里用文档形状的本地中继（http + Docker 桥接 IP），
+                # 预审走本地中继分支放行（字面 IP，不查 DNS）。
                 "_gateway_env": {"SUB2API_API_KEY": "gwkey",
-                                 "SUB2API_BASE_URL": "http://gw/"}}
+                                 "SUB2API_BASE_URL":
+                                 "http://172.17.0.1:28084"}}
         result, error = worker._run_attempt(task)
         self.assertIsNone(error)
         self.assertEqual(result["status"], "completed")

cloud/tests/test_chv_gaps.py

@@ -81,6 +81,12 @@ def _write(path, content=b"x"):
     return path
 
 
+def _public_resolver(host, port):
+    """feat-chv-muse-p9：假解析器（公网 IP），供推理出口预审用；
+    实验室绝不联网，真实解析只在真机发生。"""
+    return ["93.184.216.34"]
+
+
 class ChvGapCase(unittest.TestCase):
     def setUp(self):
         self.tmp = tempfile.TemporaryDirectory()
@@ -129,7 +135,11 @@ class ChvGapCase(unittest.TestCase):
                # genisoimage，测的不是 host key 模式，故显式
                # 声明实验室路径，绝不把它写成生产默认。
                "CHV_HOST_KEY_MODE": "keyscan",
-               "CHV_SSH_FORWARD": "1"}
+               "CHV_SSH_FORWARD": "1",
+               # feat-chv-muse-p9：推理出口预审的假 origin（公网
+               # https，配假解析器；实验室不起真 VM、不联网）。
+               "SUB2API_BASE_URL": "https://xgapi.lol",
+               "CHV_INFERENCE_ORIGIN": "https://xgapi.lol"}
         if env_extra:
             env.update(env_extra)
         kwargs = dict(
@@ -143,6 +153,7 @@ class ChvGapCase(unittest.TestCase):
             sleeper=lambda s: None, env=env)
         kwargs.update(overrides)
         ex = CloudHypervisorExecutor(**kwargs)
+        ex.sentinel_resolver = _public_resolver
         ex.audit_callback = lambda e, t, d, k=None: self.audit.append(
             (e, t.get("id") if isinstance(t, dict) else t, d, k))
         return ex

cloud/tests/test_chv_muse_p9.py

@@ -0,0 +1,570 @@
+"""对齐 Muse P9：推理与遥测受限通道。
+
+全部用标准库与 mock，不启动真实 VM、不联网、不碰 WSL/真机；
+解析器一律用假解析器（公网 IP / 私网 IP 两档）。
+"""
+
+import importlib.util
+import json
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+try:
+    from cloud.chv.executor import CloudHypervisorExecutor
+    from cloud.chv.inference_channel import (
+        InferenceLimits,
+        check_inference_call,
+        inference_allow_hosts,
+        inference_limits,
+        resolve_inference_origin,
+    )
+    from cloud.chv.sentinel import SentinelPolicy
+    from cloud.chv.telemetry import (
+        build_task_telemetry,
+        build_trajectory_event,
+        emit_telemetry,
+        sanitize_telemetry_event,
+        sampled,
+        set_training_enabled,
+        telemetry_sample_rate,
+        training_enabled,
+    )
+    from cloud.sandbox import ChvFailure
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.chv.executor import CloudHypervisorExecutor
+    from cloud.chv.inference_channel import (
+        InferenceLimits,
+        check_inference_call,
+        inference_allow_hosts,
+        inference_limits,
+        resolve_inference_origin,
+    )
+    from cloud.chv.sentinel import SentinelPolicy
+    from cloud.chv.telemetry import (
+        build_task_telemetry,
+        build_trajectory_event,
+        emit_telemetry,
+        sanitize_telemetry_event,
+        sampled,
+        set_training_enabled,
+        telemetry_sample_rate,
+        training_enabled,
+    )
+    from cloud.sandbox import ChvFailure
+
+
+def _public_resolver(host, port):
+    """假解析器：公网 IP。"""
+    return ["93.184.216.34"]
+
+
+def _private_resolver(host, port):
+    """假解析器：私网 IP（SSRF 场景）。"""
+    return ["10.0.0.5"]
+
+
+def _load_host_module():
+    """以独立模块名加载仓库根的 host.py（只用标准库）。"""
+    path = Path(__file__).resolve().parents[2] / "host.py"
+    spec = importlib.util.spec_from_file_location("host_under_test", path)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class OriginTest(unittest.TestCase):
+    def test_valid_sub2api_origin(self):
+        env = {"SUB2API_BASE_URL": "https://xgapi.lol/"}
+        self.assertEqual(
+            resolve_inference_origin("sub2api", env), "https://xgapi.lol")
+
+    def test_missing_origin_returns_empty(self):
+        self.assertEqual(resolve_inference_origin("sub2api", {}), "")
+        self.assertEqual(resolve_inference_origin(None, {}), "")
+
+    def test_http_origin_rejected(self):
+        env = {"SUB2API_BASE_URL": "http://xgapi.lol"}
+        with self.assertRaises(ValueError):
+            resolve_inference_origin("sub2api", env)
+
+    def test_http_public_ip_rejected(self):
+        env = {"SUB2API_BASE_URL": "http://8.8.8.8:28084"}
+        with self.assertRaises(ValueError):
+            resolve_inference_origin("sub2api", env)
+
+    def test_http_hostname_rejected(self):
+        env = {"SUB2API_BASE_URL": "http://relay.local:28084"}
+        with self.assertRaises(ValueError):
+            resolve_inference_origin("sub2api", env)
+
+    def test_http_local_relay_allowed(self):
+        # 文档形状：本地 sub2api 中继走 Docker 桥接 IP + http，
+        # 端口保留；环回同款。
+        env = {"SUB2API_BASE_URL": "http://172.17.0.1:28084/"}
+        self.assertEqual(
+            resolve_inference_origin("sub2api", env),
+            "http://172.17.0.1:28084")
+        env2 = {"SUB2API_BASE_URL": "http://127.0.0.1:28084"}
+        self.assertEqual(
+            resolve_inference_origin("sub2api", env2),
+            "http://127.0.0.1:28084")
+
+    def test_userinfo_in_origin_rejected(self):
+        env = {"SUB2API_BASE_URL": "https://user:pass@xgapi.lol"}
+        with self.assertRaises(ValueError):
+            resolve_inference_origin("sub2api", env)
+
+    def test_non_443_port_rejected(self):
+        env = {"SUB2API_BASE_URL": "https://xgapi.lol:8443"}
+        with self.assertRaises(ValueError):
+            resolve_inference_origin("sub2api", env)
+
+    def test_unknown_provider_falls_back_to_default_origin(self):
+        # 未知标签与 host.py `_server_command` 一致走默认网关；
+        # 未配 CHV_INFERENCE_ORIGIN 时返回 ""（调用方按 ask 处理）。
+        env = {"CHV_INFERENCE_ORIGIN": "https://xgapi.lol"}
+        self.assertEqual(
+            resolve_inference_origin("openai", env), "https://xgapi.lol")
+        self.assertEqual(resolve_inference_origin("openai", {}), "")
+
+    def test_xcmapi_origin_from_explicit_env(self):
+        env = {"CHV_INFERENCE_ORIGIN": "https://xgapi.lol"}
+        self.assertEqual(
+            resolve_inference_origin(None, env), "https://xgapi.lol")
+
+
+class AllowlistTest(unittest.TestCase):
+    def test_explicit_allowlist_wins(self):
+        env = {"SUB2API_BASE_URL": "https://xgapi.lol",
+               "CHV_INFERENCE_ALLOW_HOSTS": "a.example, B.example "}
+        hosts = inference_allow_hosts("sub2api", env)
+        self.assertEqual(hosts, frozenset({"a.example", "b.example"}))
+
+    def test_default_allowlist_is_origin_host(self):
+        env = {"SUB2API_BASE_URL": "https://xgapi.lol"}
+        self.assertEqual(inference_allow_hosts("sub2api", env),
+                         frozenset({"xgapi.lol"}))
+
+    def test_no_origin_means_empty_allowlist(self):
+        self.assertEqual(inference_allow_hosts("sub2api", {}), frozenset())
+
+
+class LimitsTest(unittest.TestCase):
+    def test_defaults(self):
+        self.assertEqual(inference_limits({}),
+                         InferenceLimits(timeout_seconds=120,
+                                         max_retries=1))
+
+    def test_env_override(self):
+        env = {"CHV_INFERENCE_TIMEOUT": "60",
+               "CHV_INFERENCE_MAX_RETRIES": "3"}
+        self.assertEqual(inference_limits(env),
+                         InferenceLimits(timeout_seconds=60, max_retries=3))
+
+    def test_invalid_timeout_rejected(self):
+        for bad in ("abc", "5", "601", "-1"):
+            with self.assertRaises(ValueError):
+                inference_limits({"CHV_INFERENCE_TIMEOUT": bad})
+
+    def test_invalid_retries_rejected(self):
+        for bad in ("x", "-1", "6"):
+            with self.assertRaises(ValueError):
+                inference_limits({"CHV_INFERENCE_MAX_RETRIES": bad})
+
+
+class PreflightTest(unittest.TestCase):
+    def setUp(self):
+        self.env = {"SUB2API_BASE_URL": "https://xgapi.lol",
+                    "CHV_INFERENCE_ALLOW_HOSTS": "xgapi.lol"}
+
+    def test_allow_with_public_ip(self):
+        check = check_inference_call(
+            "sub2api", self.env, resolve=_public_resolver)
+        self.assertEqual(check.decision.verdict, "allow")
+        self.assertIsNotNone(check.decision.capability)
+        self.assertTrue(check.decision.capability.matches(check.request))
+        self.assertEqual(check.request.host, "xgapi.lol")
+        self.assertEqual(check.request.path, "/v1/responses")
+
+    def test_capability_single_use_300s(self):
+        check = check_inference_call(
+            "sub2api", self.env, resolve=_public_resolver, now=1000.0)
+        cap = check.decision.capability
+        self.assertTrue(cap.use(check.request, now=1000.0))
+        self.assertFalse(cap.use(check.request, now=1000.0))
+        self.assertFalse(cap.matches(check.request, now=2000.0))
+
+    def test_local_relay_allow_without_dns(self):
+        # 本地中继是字面私网 IP：不查 DNS（解析器被调即错），
+        # allowlist 默认即 origin 主机，直接 allow + 单次能力。
+        def _must_not_resolve(host, port):
+            raise AssertionError("local relay must not resolve")
+        env = {"SUB2API_BASE_URL": "http://172.17.0.1:28084"}
+        check = check_inference_call(
+            "sub2api", env, resolve=_must_not_resolve)
+        self.assertEqual(check.decision.verdict, "allow")
+        self.assertIn("local_relay", check.decision.reasons)
+        self.assertIsNotNone(check.decision.capability)
+        self.assertEqual(check.request.host, "172.17.0.1")
+        self.assertEqual(check.request.port, 28084)
+        self.assertEqual(check.request.scheme, "http")
+
+    def test_local_relay_ask_when_not_allowlisted(self):
+        env = {"SUB2API_BASE_URL": "http://172.17.0.1:28084",
+               "CHV_INFERENCE_ALLOW_HOSTS": "xgapi.lol"}
+        check = check_inference_call(
+            "sub2api", env, resolve=_public_resolver)
+        self.assertEqual(check.decision.verdict, "ask")
+
+    def test_deny_on_private_ip(self):
+        check = check_inference_call(
+            "sub2api", self.env, resolve=_private_resolver)
+        self.assertEqual(check.decision.verdict, "deny")
+        self.assertIn("ssrf_private_ip", check.decision.reasons[0])
+
+    def test_ask_on_non_allowlisted_host(self):
+        # 显式 allowlist 已配、origin 主机不在其中 → ask。
+        # （未配 allowlist 时默认即 origin 主机自身，不会 ask——
+        # 见 test_default_allowlist_is_origin_host。）
+        env = {"SUB2API_BASE_URL": "https://unknown.example",
+               "CHV_INFERENCE_ALLOW_HOSTS": "xgapi.lol"}
+        check = check_inference_call(
+            "sub2api", env, resolve=_public_resolver)
+        self.assertEqual(check.decision.verdict, "ask")
+
+    def test_ask_when_origin_not_configured(self):
+        check = check_inference_call(
+            None, {}, resolve=_public_resolver)
+        self.assertEqual(check.decision.verdict, "ask")
+        self.assertIn("inference_origin_not_configured",
+                      check.decision.reasons)
+
+    def test_xcmapi_origin_allow(self):
+        env = {"CHV_INFERENCE_ORIGIN": "https://xgapi.lol",
+               "CHV_INFERENCE_ALLOW_HOSTS": "xgapi.lol"}
+        check = check_inference_call(
+            None, env, resolve=_public_resolver)
+        self.assertEqual(check.decision.verdict, "allow")
+
+    def test_unknown_provider_falls_back_to_default_origin(self):
+        # 未知标签不 deny：与 _server_command 一致走默认网关；
+        # 未配 CHV_INFERENCE_ORIGIN 时仍是 ask（fail-closed）。
+        env = {"CHV_INFERENCE_ORIGIN": "https://xgapi.lol",
+               "CHV_INFERENCE_ALLOW_HOSTS": "xgapi.lol"}
+        check = check_inference_call(
+            "prov-x", env, resolve=_public_resolver)
+        self.assertEqual(check.decision.verdict, "allow")
+        check2 = check_inference_call(
+            "prov-x", {}, resolve=_public_resolver)
+        self.assertEqual(check2.decision.verdict, "ask")
+
+    def test_deny_path_out_of_scope(self):
+        policy = SentinelPolicy(
+            allow_hosts={"xgapi.lol"},
+            connector_paths={"inference": (("POST", "prefix",
+                                            "/v1/responses"),)})
+        from cloud.chv.sentinel import EgressRequest
+        request = EgressRequest("inference", "GET", "https",
+                                "xgapi.lol", 443, "/v1/admin")
+        decision = policy.evaluate(request, resolve=_public_resolver)
+        self.assertEqual(decision.verdict, "deny")
+
+    def test_audit_record_has_no_values(self):
+        check = check_inference_call(
+            "sub2api", self.env, resolve=_public_resolver)
+        record = check.decision.audit(check.request)
+        blob = json.dumps(record)
+        self.assertNotIn("xgapi.lol".replace("xgapi", "SECRET"), blob)
+        self.assertEqual(record["host"], "xgapi.lol")
+        self.assertEqual(record["verdict"], "allow")
+
+
+class FakeResult:
+    def __init__(self, returncode=0, stdout="", stderr=""):
+        self.returncode = returncode
+        self.stdout = stdout
+        self.stderr = stderr
+
+
+def _write(path, content=b"x"):
+    with open(path, "wb") as fh:
+        fh.write(content)
+    os.chmod(path, 0o755)
+    return path
+
+
+class GateTest(unittest.TestCase):
+    """执行器预审：deny/ask 时不起 VM（wake 未被调用）；allow 才继续。"""
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.dir = self.tmp.name
+        self.binary = _write(os.path.join(self.dir, "cloud-hypervisor"))
+        self.ch_remote = _write(os.path.join(self.dir, "ch-remote"))
+        self.kernel = _write(os.path.join(self.dir, "vmlinux"))
+        self.rootfs = _write(os.path.join(self.dir, "disk.raw"), b"R" * 100)
+        self.ssh_key = _write(os.path.join(self.dir, "id_chv"))
+        self.kvm = _write(os.path.join(self.dir, "kvm"))
+        self.cpuinfo = os.path.join(self.dir, "cpuinfo")
+        with open(self.cpuinfo, "w") as fh:
+            fh.write("flags : fpu vmx sse\n")
+        self.state_dir = os.path.join(self.dir, "state")
+        self.wake_calls = []
+        self.audit = []
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _make_exec(self, env_extra):
+        env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888",
+               "CHV_HOST_KEY_MODE": "keyscan",
+               "CHV_SSH_FORWARD": "1"}
+        env.update(env_extra)
+        ex = CloudHypervisorExecutor(
+            binary=self.binary, ch_remote=self.ch_remote,
+            kernel=self.kernel, rootfs=self.rootfs,
+            ssh_key=self.ssh_key, kvm_path=self.kvm,
+            cpuinfo_path=self.cpuinfo, state_dir=self.state_dir,
+            runner=lambda cmd, timeout=None: FakeResult(0, "", ""),
+            launcher=lambda cmd: FakeResult(0, "", ""),
+            api_client=lambda *a, **k: FakeResult(0, "", ""),
+            sleeper=lambda s: None, env=env)
+        ex.audit_callback = lambda e, t, d, k=None: self.audit.append(
+            (e, t.get("id") if isinstance(t, dict) else t, d, k))
+        return ex
+
+    def test_deny_blocks_wake(self):
+        ex = self._make_exec(
+            {"SUB2API_BASE_URL": "https://xgapi.lol",
+             "CHV_INFERENCE_ALLOW_HOSTS": "xgapi.lol"})
+        ex.sentinel_resolver = _private_resolver
+        ex.wake_vm = mock.Mock(side_effect=AssertionError("must not wake"))
+        task = {"id": "t1", "prompt": "hi"}
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.dispatch_task(task, model_provider="sub2api")
+        self.assertEqual(ctx.exception.kind, "inference_egress_denied")
+        ex.wake_vm.assert_not_called()
+        kinds = [e[0] for e in self.audit]
+        self.assertIn("inference_egress_deny", kinds)
+
+    def test_ask_blocks_wake_when_origin_missing(self):
+        ex = self._make_exec({})
+        ex.sentinel_resolver = _public_resolver
+        ex.wake_vm = mock.Mock(side_effect=AssertionError("must not wake"))
+        task = {"id": "t2", "prompt": "hi"}
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.dispatch_task(task, model_provider=None)
+        self.assertEqual(ctx.exception.kind, "inference_egress_denied")
+        ex.wake_vm.assert_not_called()
+
+    def test_extra_env_origin_is_honored(self):
+        # 网关配置随任务来（worker 的 extra_env），执行器
+        # 进程 env 里没有 origin 也要放行。
+        ex = self._make_exec({})
+        ex.sentinel_resolver = _public_resolver
+        ex.wake_vm = mock.Mock()
+        task = {"id": "t3", "prompt": "hi"}
+        with mock.patch.object(ex, "_run_guest",
+                               return_value={"status": "completed"}):
+            with mock.patch.object(ex, "shutdown_vm"):
+                ex.dispatch_task(
+                    task, model_provider="sub2api",
+                    extra_env={"SUB2API_BASE_URL": "https://xgapi.lol"})
+        ex.wake_vm.assert_called_once()
+
+    def test_bad_origin_becomes_deny_not_value_error(self):
+        ex = self._make_exec({})
+        ex.sentinel_resolver = _public_resolver
+        ex.wake_vm = mock.Mock(side_effect=AssertionError("must not wake"))
+        task = {"id": "t4", "prompt": "hi"}
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.dispatch_task(
+                task, model_provider="sub2api",
+                extra_env={"SUB2API_BASE_URL": "http://xgapi.lol"})
+        self.assertEqual(ctx.exception.kind, "inference_egress_denied")
+        ex.wake_vm.assert_not_called()
+        kinds = [e[0] for e in self.audit]
+        self.assertIn("inference_egress_deny", kinds)
+
+
+class TelemetryTest(unittest.TestCase):
+    def test_allowlisted_fields_only(self):
+        event = build_task_telemetry(
+            task_id="t1", vm_id="vm1", provider="sub2api",
+            model="gpt-6.1-terra", status="completed",
+            started_at=1000.0, ended_at=1005.0, duration_ms=5000,
+            tokens={"input_tokens": 10, "output_tokens": 5,
+                    "total_tokens": 15},
+            turns=2, tool_calls=3, env={"CHV_TELEMETRY_SAMPLE_RATE": "1"})
+        self.assertIsNotNone(event)
+        self.assertEqual(event["tokens_in"], 10)
+        self.assertEqual(event["tokens_total"], 15)
+        self.assertEqual(event["turns"], 2)
+
+    def test_secret_and_text_keys_dropped(self):
+        event = {"task_id": "t1", "vm_id": "v", "provider": "sub2api",
+                 "model": "m", "status": "completed",
+                 "started_at": 1.0, "ended_at": 2.0, "duration_ms": 1,
+                 "prompt": "do evil", "content": "secret file",
+                 "api_key": "sk-123", "user_email": "a@b.c",
+                 "extra_blob": {"nested": "data"}}
+        clean = sanitize_telemetry_event(event)
+        for bad in ("prompt", "content", "api_key", "user_email",
+                    "extra_blob"):
+            self.assertNotIn(bad, clean)
+        self.assertEqual(clean["task_id"], "t1")
+
+    def test_tokens_none_when_not_reported(self):
+        event = build_task_telemetry(
+            task_id="t1", vm_id="vm1", provider="sub2api", model=None,
+            status="failed", started_at=1.0, ended_at=2.0, duration_ms=1,
+            env={"CHV_TELEMETRY_SAMPLE_RATE": "1"})
+        self.assertIsNone(event["tokens_in"])
+        self.assertIsNone(event["tokens_out"])
+        self.assertIsNone(event["tokens_total"])
+
+    def test_sampling_deterministic(self):
+        self.assertEqual(sampled("task-abc", 1.0), True)
+        self.assertEqual(sampled("task-abc", 0.0), False)
+        self.assertEqual(sampled("task-abc", 0.5),
+                         sampled("task-abc", 0.5))
+        event = build_task_telemetry(
+            task_id="t1", vm_id="v", provider="sub2api", model=None,
+            status="completed", started_at=1.0, ended_at=2.0,
+            duration_ms=1, env={"CHV_TELEMETRY_SAMPLE_RATE": "0"})
+        self.assertIsNone(event)
+
+    def test_invalid_sample_rate_rejected(self):
+        with self.assertRaises(ValueError):
+            telemetry_sample_rate({"CHV_TELEMETRY_SAMPLE_RATE": "2"})
+        with self.assertRaises(ValueError):
+            telemetry_sample_rate({"CHV_TELEMETRY_SAMPLE_RATE": "x"})
+
+    def test_emit_never_raises(self):
+        self.assertFalse(emit_telemetry("/nonexistent/dir/xyz",
+                                        {"task_id": "t1"}))
+        self.assertFalse(emit_telemetry("", {"task_id": "t1"}))
+
+    def test_emit_writes_jsonl(self):
+        with tempfile.TemporaryDirectory() as d:
+            event = {"task_id": "t9", "vm_id": "v", "status": "completed"}
+            self.assertTrue(emit_telemetry(d, event))
+            with open(os.path.join(d, "telemetry.jsonl")) as fh:
+                row = json.loads(fh.read().strip())
+            self.assertEqual(row["task_id"], "t9")
+            self.assertNotIn("prompt", row)
+
+
+class TrainingOptTest(unittest.TestCase):
+    def test_default_enabled(self):
+        with tempfile.TemporaryDirectory() as d:
+            self.assertTrue(training_enabled(d))
+
+    def test_set_off_persists_and_blocks_trajectory(self):
+        with tempfile.TemporaryDirectory() as d:
+            self.assertTrue(set_training_enabled(d, False))
+            self.assertFalse(training_enabled(d))
+            event = build_trajectory_event(
+                "t1", 3, ["read", "write"], 100, d)
+            self.assertIsNone(event)
+            # 任务级遥测不受开关影响
+            task_event = build_task_telemetry(
+                "t1", "v", "sub2api", None, "completed",
+                1.0, 2.0, 1, env={"CHV_TELEMETRY_SAMPLE_RATE": "1"})
+            self.assertIsNotNone(task_event)
+
+    def test_trajectory_has_metadata_only(self):
+        with tempfile.TemporaryDirectory() as d:
+            event = build_trajectory_event(
+                "t1", 2, ["read", "exec"], 50, d)
+            self.assertIsNotNone(event)
+            self.assertEqual(event["tools"], ["read", "exec"])
+            self.assertNotIn("prompt", event)
+            self.assertNotIn("content", event)
+
+    def test_corrupt_state_fails_closed(self):
+        with tempfile.TemporaryDirectory() as d:
+            with open(os.path.join(d, "training-opt.json"), "w") as fh:
+                fh.write("{not json")
+            self.assertFalse(training_enabled(d))
+
+    def test_non_bool_rejected(self):
+        with tempfile.TemporaryDirectory() as d:
+            with self.assertRaises(TypeError):
+                set_training_enabled(d, "off")
+
+    def test_switch_change_is_audited(self):
+        with tempfile.TemporaryDirectory() as d:
+            set_training_enabled(d, False, actor="tester")
+            audit_path = os.path.join(d, "training-opt-audit.jsonl")
+            with open(audit_path) as fh:
+                row = json.loads(fh.read().strip())
+            self.assertEqual(row["event"], "training_opt_changed")
+            self.assertFalse(row["enabled"])
+            self.assertEqual(row["actor"], "tester")
+
+
+class HostPyTest(unittest.TestCase):
+    def setUp(self):
+        self.host = _load_host_module()
+
+    def test_server_command_valid_origin(self):
+        with mock.patch.dict(os.environ,
+                             {"SUB2API_BASE_URL": "https://xgapi.lol"}):
+            cmd = self.host._server_command("codex", provider="sub2api")
+        self.assertIn("app-server", cmd)
+        self.assertTrue(any("xgapi.lol/v1" in c for c in cmd))
+
+    def test_server_command_http_origin_rejected(self):
+        with mock.patch.dict(os.environ,
+                             {"SUB2API_BASE_URL": "http://xgapi.lol"}):
+            with self.assertRaises(ValueError):
+                self.host._server_command("codex", provider="sub2api")
+
+    def test_server_command_local_relay_allowed(self):
+        # 文档形状的本地中继（http + 桥接 IP）必须通过第二道防线。
+        with mock.patch.dict(os.environ,
+                             {"SUB2API_BASE_URL":
+                              "http://172.17.0.1:28084"}):
+            cmd = self.host._server_command("codex", provider="sub2api")
+        self.assertTrue(any("172.17.0.1:28084/v1" in c for c in cmd))
+
+    def test_server_command_allowlist_mismatch_rejected(self):
+        with mock.patch.dict(os.environ,
+                             {"SUB2API_BASE_URL": "https://xgapi.lol",
+                              "CHV_INFERENCE_ALLOW_HOSTS": "other.example"}):
+            with self.assertRaises(ValueError):
+                self.host._server_command("codex", provider="sub2api")
+
+    def test_request_timeout_default_and_env(self):
+        self.assertEqual(
+            self.host._inference_request_timeout(None), 120)
+        self.assertEqual(
+            self.host._inference_request_timeout(60), 60)
+        with mock.patch.dict(os.environ,
+                             {"CHV_INFERENCE_TIMEOUT": "30"}):
+            self.assertEqual(
+                self.host._inference_request_timeout(None), 30)
+        with mock.patch.dict(os.environ,
+                             {"CHV_INFERENCE_TIMEOUT": "5"}):
+            with self.assertRaises(ValueError):
+                self.host._inference_request_timeout(None)
+
+    def test_request_retries_default_and_env(self):
+        self.assertEqual(self.host._inference_max_retries(), 1)
+        with mock.patch.dict(os.environ,
+                             {"CHV_INFERENCE_MAX_RETRIES": "0"}):
+            self.assertEqual(self.host._inference_max_retries(), 0)
+        with mock.patch.dict(os.environ,
+                             {"CHV_INFERENCE_MAX_RETRIES": "9"}):
+            with self.assertRaises(ValueError):
+                self.host._inference_max_retries()
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_chv_optimize.py

@@ -86,6 +86,12 @@ def _write(path, content=b"x"):
     return path
 
 
+def _public_resolver(host, port):
+    """feat-chv-muse-p9：假解析器（公网 IP），供推理出口预审用；
+    实验室绝不联网，真实解析只在真机发生。"""
+    return ["93.184.216.34"]
+
+
 class ChvOptCase(unittest.TestCase):
     def setUp(self):
         self.tmp = tempfile.TemporaryDirectory()
@@ -118,7 +124,11 @@ class ChvOptCase(unittest.TestCase):
                # genisoimage，测的不是 host key 模式，故显式
                # 声明实验室路径，绝不把它写成生产默认。
                "CHV_HOST_KEY_MODE": "keyscan",
-               "CHV_SSH_FORWARD": "1"}
+               "CHV_SSH_FORWARD": "1",
+               # feat-chv-muse-p9：推理出口预审的假 origin（公网
+               # https，配假解析器；实验室不起真 VM、不联网）。
+               "SUB2API_BASE_URL": "https://xgapi.lol",
+               "CHV_INFERENCE_ORIGIN": "https://xgapi.lol"}
         if env_extra:
             env.update(env_extra)
         if runner is None:
@@ -135,7 +145,9 @@ class ChvOptCase(unittest.TestCase):
             runner=runner, launcher=launcher, api_client=api_client,
             sleeper=lambda s: None, env=env)
         kwargs.update(overrides)
-        return CloudHypervisorExecutor(**kwargs)
+        ex = CloudHypervisorExecutor(**kwargs)
+        ex.sentinel_resolver = _public_resolver
+        return ex
 
     def _default_runner(self, cmd, timeout=None):
         self.calls.append(list(cmd))

cloud/tests/test_cloud_hypervisor.py

@@ -73,6 +73,12 @@ def _write(path, content=b"x"):
     return path
 
 
+def _public_resolver(host, port):
+    """feat-chv-muse-p9：假解析器（公网 IP），供推理出口预审用；
+    实验室绝不联网，真实解析只在真机发生。"""
+    return ["93.184.216.34"]
+
+
 class ChvCase(unittest.TestCase):
     def setUp(self):
         self.tmp = tempfile.TemporaryDirectory()
@@ -106,7 +112,11 @@ class ChvCase(unittest.TestCase):
                # genisoimage，测的不是 host key 模式，故显式
                # 声明实验室路径，绝不把它写成生产默认。
                "CHV_HOST_KEY_MODE": "keyscan",
-               "CHV_SSH_FORWARD": "1"}
+               "CHV_SSH_FORWARD": "1",
+               # feat-chv-muse-p9：推理出口预审的假 origin（公网
+               # https，配假解析器；实验室不起真 VM、不联网）。
+               "SUB2API_BASE_URL": "https://xgapi.lol",
+               "CHV_INFERENCE_ORIGIN": "https://xgapi.lol"}
         if env_extra:
             env.update(env_extra)
         if runner is None:
@@ -124,6 +134,7 @@ class ChvCase(unittest.TestCase):
             sleeper=lambda s: None, env=env)
         kwargs.update(overrides)
         ex = CloudHypervisorExecutor(**kwargs)
+        ex.sentinel_resolver = _public_resolver
         ex.audit_callback = lambda e, t, d, k=None: self.audit.append(
             (e, t.get("id") if isinstance(t, dict) else t, d, k))
         return ex

docs/chv-access-boundary.md

@@ -0,0 +1,45 @@
+# CHV VM 运营访问边界说明（feat-chv-muse-p9）
+
+> 口径依据：Meta 官方安全文（security.muse.ai 2026-09-08）对 Muse 当前架构的
+> 明确表述——当前架构**通过运营政策限制 Meta 人员访问用户数据，但并不阻止
+> Meta 在必要时为支持、保障或运营服务而访问数据**；“连 Meta 也看不到”是未来
+> Confidential VM 的目标，**截至 2026-11-01 的公开材料中未见 Confidential VM
+> 已全面上线**。本文件把同一口径落到本项目：只写文档 + 审计钩子，
+> **不冒充技术上不可访问**。
+
+## 1. 谁能访问 VM 内数据
+
+| 访问方 | 能访问什么 | 边界 |
+|---|---|---|
+| 用户本人（VM 属主） | 全部：工作区文件、记忆、凭证替身、审计记录 | 属主即所有者；可检查、编辑、下载 VM 内所有文件，也可要求忘记指定内容（见 P7） |
+| Agent（codex，单元内） | 任务所需的工作区与工具输出 | 受 nspawn 单元（P2）、authd 替身凭证（P3）、Sentinel 出口审批（P3/P5）约束；真凭证与宿主侧数据对单元内不可见 |
+| 宿主运维（部署机器的 root / codexsvc） | VM 磁盘镜像、state_dir、日志、审计 | **技术上可以访问**：宿主侧持有磁盘文件与进程，OS 账户即边界。本文件不声称运维看不到——只规定运维**应在什么情况下**访问（见 §2），并要求访问留审计 |
+| Meta（对照口径） | 不适用——本项目无 Meta 侧运营方 | 本项目是用户自部署，不存在 Meta 人员访问；写这一行只为对齐口径：若未来托管化，边界按同一句话执行 |
+
+## 2. 运维访问规则（政策，非技术强制）
+
+1. **日常不看**：运维不主动翻看用户工作区文件、对话记录与记忆；只看任务级元数据（任务 id、起止、成败、耗时）与审计事件。
+2. **排障可看**：任务失败、VM 异常、安全事件调查时，运维可查看相关 VM 的日志、审计记录与任务输出；查看行为本身记入审计（`chv_audit` + 操作日志）。
+3. **凭证永不直接取**：真凭证（Gitea token、网关 key）只经 authd 替身释放；运维排障需要凭证时走替身铸发流程，不直接读 seed 文件；seed 文件 0600，属主 root/codexsvc。
+4. **数据不出宿主**：排障取证不出部署机器；必须外发时先去标识化（任务 id 可留，对话原文、文件内容、密钥一律去掉），并记审计。
+5. **删除即删除**：用户要求删除任务数据/记忆时，运维删除 VM 盘、state_dir 与相关遥测行，并记审计；备份里的副本按备份保留期滚动清除，不单点恢复。
+
+## 3. 审计（可查的证据）
+
+- 任务生命周期审计：`chv_audit`（执行器 `_emit`，失败落盘 `chv-audit-spill.jsonl`）。
+- 出口审批审计：Sentinel 决策记录（`sentinel_egress_allow/deny/ask`，只含目标描述与原因，不含值）。
+- 训练开关审计：`training-opt-audit.jsonl`（开关变更时间、开/关、操作方）。
+- 遥测：`telemetry.jsonl`（只含任务级元数据字段，见 `docs/chv-muse-p9-gap.md` §2.1）。
+
+审计记录里绝不含凭证值、对话原文与文件内容；审计失败不中断任务（worker 层 best-effort 语义，见执行器 `_emit` 注释）。
+
+## 4. 广告隔离（口径，不用写代码）
+
+- 本项目不把用户对话或 VM 数据分享给任何广告系统；遥测字段白名单里没有可用于广告画像的字段。
+- Agent 浏览互联网时表现为用户活动，可能间接影响用户在别处看到的广告（与 Muse 官方口径一致）——这是浏览行为本身的副作用，不是本项目分享数据。
+
+## 5. 明确未做的事
+
+- **Confidential VM**：未做。Meta 自己也未交付（路线图），本项目不自研密码学隔离 VM；当前边界是 OS 账户 + 文件权限 + 审计，不是密码学不可访问。任何“连运维也看不到”的说法在本架构下都是错的，不要这样对外写。
+- **端到端加密记忆**：未做。记忆文件在 VM 盘上明文存放（权限 0600），依赖磁盘与宿主访问控制。
+- **操作审批 UI**：运维访问的审批流程目前是人工约定（本文件），没有强制审批卡；强制审批是后续工作，不在本轨冒充已实现。

docs/chv-muse-p9-gap.md

@@ -0,0 +1,157 @@
+# 对齐 Muse P9 差距表：推理与遥测受限通道
+
+> 先写表再动手。依据：`~/workspace/research_notes/meta-muse-ai-architecture-20261101-0755/report.md`
+> （Meta 官方 security.muse.ai 与设计文 2026-09-08 为准；第三方拆解只作参考，不得把逆向细节写成官方架构）。
+> 基线：`06c3e17`。分支：`feat-chv-muse-p9`。本地做，不推、不合、不部署。
+
+## 1. 现状 vs Muse 官方要求
+
+| # | Muse 官方要求（security.muse.ai / 设计文） | 现状（基线 06c3e17） | 本轨做法 |
+|---|---|---|---|
+| 1 | 推理与遥测代理是通向外部基础设施的**受限通道** | 模型调用无收敛点：`host.py` 起 `codex app-server`，provider=`sub2api` 时 `-c` 覆盖 base_url（取 `SUB2API_BASE_URL`）；provider 为空时 codex 用自身 config.toml 默认网关（xcmapi 路径），执行器侧没有任何出口校验 | 新模块 `cloud/chv/inference_channel.py`：单一出口点 + 执行器 `_run_with_vm` 起 VM 前预审（Sentinel 评估）；`host.py` 的 `_server_command` 做结构校验（https、无 userinfo、主机在 allowlist 内）当第二道防线 |
+| 2 | VM 发送**有限数据**用于推理与遥测（安全文**未给字段级清单**） | 无任何遥测：任务元数据只进宿主日志与审计 spill，无字段清单、无采样、无发送路径 | 新模块 `cloud/chv/telemetry.py`：字段级白名单（只含任务级元数据，见 §2）；采样率可配；落盘 `<state_dir>/telemetry.jsonl`，发送失败只记日志、不影响任务 |
+| 3 | 推理轨迹（对话往返、工具调用、子代理交接）在**去除关键个人身份信息后**可用于训练新检查点；用户可**一键退出训练** | 无训练上传、无开关 | 训练退出开关（`training_opt`，见 §2）：状态持久化在 `<state_dir>/training-opt.json`，关后轨迹类事件不再产生；开关变更进审计。**默认值开是本项目决定，非 Meta 官方值**（报告注明帮助中心原文未逐字确认默认开启） |
+| 4 | 当前 VM 的 Meta 访问边界（官方明确，非营销口径）：运营政策限制 Meta 人员访问用户数据，但**并不阻止** Meta 在必要时为支持、保障或运营而访问；“连 Meta 也看不到”是未来 Confidential VM 的目标 | 无边界说明 | `docs/chv-access-boundary.md`：只写文档 + 审计钩子，不冒充技术上不可访问；Confidential VM 明确列为未做 |
+| 5 | 广告隔离：不把用户对话或 VM 数据分享给广告系统；但浏览互联网时表现为用户活动、可能间接影响广告 | 无口径 | 只记为口径（见 §4），不用写代码 |
+| 6 | Sentinel 是连接器动作与全部网络出口的唯一权限机构 | P3 已做 Sentinel 最小审批、P5 已做纵深（本基线**都没有**；P3 在 `feat-chv-muse-p3` 分支 a90cf78，P5 在 `feat-chv-muse-p5` 分支 25bcd35） | 推理通道走 Sentinel 审批，不重复造审批：`cloud/chv/sentinel.py` 从 **P5 的 25bcd35 拷入**（P5 版已是 P3 的超集，顶层仍只用标准库），再加 P9 的 inference 条目（三张表里加 `inference`：scheme https、端口 443、路径 `POST /v1/responses` + `GET /v1/models`）；合入时以本文件为准（它是 P3+P5+P9 的并集） |
+
+## 2. 本项目自定清单（非 Meta 官方值，逐项注明）
+
+### 2.1 遥测字段白名单（`telemetry.py`）
+
+只含任务级元数据，**绝不含**对话原文、文件内容、密钥、个人身份信息：
+
+| 字段 | 含义 | 来源 |
+|---|---|---|
+| `task_id` | 任务 id | 任务 dict |
+| `vm_id` | VM 名 | 执行器 |
+| `provider` | 模型 provider（`sub2api` / `xcmapi`） | 通道决策 |
+| `model` | 模型名 | 任务 dict |
+| `status` | `completed` / `failed` / `interrupted` / `timeout` | 运行结果 |
+| `started_at` / `ended_at` | 起止时间戳（秒） | 执行器 |
+| `duration_ms` | 耗时毫秒 | 执行器 |
+| `tokens_in` / `tokens_out` / `tokens_total` | token 数 | host.py 回传的 usage（未回传时记 `null`，**不编造**） |
+| `turns` | 回合数 | 运行结果 |
+| `tool_calls` | 工具调用次数 | 运行结果 |
+| `sampled` | 是否被采样命中 | 采样器 |
+
+不在白名单里的键一律丢弃；键名含 `key` / `token` / `secret` / `password` / `prompt` /
+`content` / `text` / `message` / `email` / `name` 的一律丢弃（防误带）。采样率
+`CHV_TELEMETRY_SAMPLE_RATE`：0.0–1.0，默认 `1.0`（全量，任务级元数据量小）；按
+`task_id` 确定性采样（同一任务每次判定一致，可测试）。
+
+### 2.2 轨迹事件（训练用）
+
+轨迹类事件只含回合级元数据（回合数、工具名、耗时），**不含任何文本**；去标识化
+= 只出元数据、不出内容。退出开关关后，轨迹类事件**不再产生**（不是产生后丢弃）。
+
+### 2.3 推理通道上限（自定）
+
+| 项 | 环境变量 | 默认 | 范围 | 说明 |
+|---|---|---|---|---|
+| 单次请求超时（秒） | `CHV_INFERENCE_TIMEOUT` | `120` | 10–600 | 与 `host.py` 的 `Server.request` 默认超时一致；非法值即 ValueError |
+| 重试上限（次） | `CHV_INFERENCE_MAX_RETRIES` | `1` | 0–5 | `Server.request` 的 BrokenPipe 重连次数上限；默认 `1` 即**保持现有行为不变**（现为 `range(2)`）；模型回合级重试仍是 worker 的 `max_retries` 语义，本轨不动 |
+
+### 2.4 训练退出开关默认值
+
+默认**开**（轨迹可用于训练）。这是**本项目决定，非 Meta 官方值**：报告注明帮助中心原文未逐字确认
+默认开启。开关状态持久化、变更进审计；关后立即生效（后续轨迹不再产生）。
+
+### 2.5 推理出口 allowlist 与 origin 规则
+
+- provider=`sub2api`：origin 取 `SUB2API_BASE_URL`（去尾斜杠）；未知 provider 标签与
+  provider 为空一样走默认网关（与 host.py `_server_command` 一致——标签本身不决定
+  origin，origin 永远只取 operator 配置的 env）。
+- provider 为空或未知标签（xcmapi 路径，codex 用自身 config.toml 默认网关）：执行器
+  **看不到** origin，改取 `CHV_INFERENCE_ORIGIN`（operator 显式配置，如 `https://xgapi.lol`）；
+  未配即 `ask`——审计 + 任务停止（fail-closed），**不静默放行**。部署迁移：现网 xcmapi 部署
+  须补设 `CHV_INFERENCE_ORIGIN`，否则任务会在预审处停下（这是有意行为，见 §3）。
+- scheme 规则（2026-11-07 全量套件复核时修正——初版只认 https 是错的）：
+  - https：公网网关（如 xgapi.lol），端口只接受 443（或缺省）；
+  - http：**只接受本地中继**——字面环回/私网 IP 主机（如 cloud/README.md 的 sub2api 网关
+    文档形状 `http://172.17.0.1:28084`，Docker 桥接 IP），端口保留；http + 公网主机名/IP、
+    http + 主机名一律拒绝。主机名一律不算本地中继（DNS 结果不可信）。
+  - 本地中继分支：字面 IP 无 DNS、无重绑定，跳过 Sentinel 的 DNS/SSRF 步；allowlist 与
+    单次能力（P3 默认档）照常执行；解析器不被调用（测试用必错解析器钉住）。Sentinel 表里的
+    inference scheme/端口仍只写 https/443——本地中继不走 `evaluate`，不要“顺手”把 http
+    加进表（那会放宽公网路径）。
+- 路径规则（`inference` 连接器）：只放 `POST /v1/responses` 与 `GET /v1/models`；
+  `/v1/models` 是否必需**待真机核实**（codex 启动时是否拉模型列表），若非必需则收紧为只放
+  `/v1/responses`。公网路径：解析到私网 IP 即 deny（SSRF，复用 Sentinel）；解析失败即 deny。
+- 任务级 env 合并：门的 origin 取任务级 `extra_env`（网关配置随任务来，如 worker 的
+  `_gateway_env`）覆盖执行器进程 env；origin 非法（http 公网、带 userinfo、https 非 443
+  端口）记 deny，不抛裸 ValueError。报错/审计原因不带 origin 原值（http URL 里可能含
+  userinfo，见 `resolve_inference_origin`）。
+
+## 3. 改什么（文件清单）
+
+| 文件 | 动作 |
+|---|---|
+| `cloud/chv/sentinel.py` | **新增**：从 P5 分支 `25bcd35` 拷入（P5 版已是 P3 的超集，顶层仍只用标准库），再加 P9 的 inference 条目（三张表：路径、scheme、端口）；合入时以本文件为准（P3+P5+P9 的并集） |
+| `cloud/chv/inference_channel.py` | **新增**：受限推理通道（origin 解析、allowlist、上限校验、Sentinel 预审、`InferenceChannelDenied` 之外的 Decision 返回） |
+| `cloud/chv/telemetry.py` | **新增**：遥测事件（字段白名单、采样、落盘、永不抛错）+ 训练退出开关（持久化、审计） |
+| `cloud/chv/executor.py` | **改动**：`_run_with_vm` 起 VM 前调推理预审（`_inference_gate`）；allow 通过、deny/ask 审计 + `ChvFailure` 停止，不起 VM。`execute()`（只读 status，不跑模型回合）不设门 |
+| `host.py` | **改动**：`_server_command` 的 sub2api origin 结构校验（https、无 userinfo、主机在 allowlist 内）；`Server.request` 默认超时与重连上限改走 `CHV_INFERENCE_TIMEOUT` / `CHV_INFERENCE_MAX_RETRIES`（默认 120 秒 / 1 次，即现有行为） |
+| `cloud/tests/test_chv_muse_p9.py` | **新增**：全部覆盖（见 §5） |
+| `docs/chv-muse-p9-gap.md` | **新增**：本文件 |
+| `docs/chv-access-boundary.md` | **新增**：运营访问边界说明 |
+
+**不碰的**：`cloud/chv/config.py`（通道读 `env` 字典，不进 VM 配置片段；架构测试无“env 必须进片段”的规矩）、
+P3 的 authd、P5 的 eBPF 污点（P5 已结论：本 VM 的 bpf() 被 seccomp 挡，不可行；本轨不重复探）、
+P6 的确定性检查层（审计复用执行器已有的 `_emit`，不另造审计接口）、worker 的 `max_retries`、
+guest 镜像（`host.py` 是标准库单文件，`/app/host.py` 拷入方式不变——本轨对 `host.py` 的改动只用标准库）。
+
+## 4. 诚实边界
+
+- 本轨是**决策层 + 预审**，不是网络层强制：客内 codex 直连的强制仍靠 Tinyproxy 白名单 + 宿主防火墙
+  （第三棒落地中）。不要把本轨写成“出口已全部经 Sentinel”——P3 的模块 docstring 同款声明。
+- provider 为空时执行器看不到 codex config.toml 里的 origin：本轨用 `CHV_INFERENCE_ORIGIN`
+  显式配置 + 未配即 ask 兜底；这不是“读到了 config”，部署时须补 env（见 §2.5）。
+- 遥测目前只落盘 `<state_dir>/telemetry.jsonl`，**没有**向外部遥测端点发送的路径；“发送失败不影响任务”
+  指落盘失败也不影响任务。向外部发遥测是后续工作，需另批。
+- 训练轨迹目前**没有**上传路径：开关管的是“轨迹类事件是否产生”；上传通道本身不存在，开关是为
+  上传通道预留的门禁，不是已接通的上传。
+- token 数只取 host.py 回传的 usage；codex 未回传时记 `null`，不估算、不编造。
+- `GET /v1/models` 是否必需未验证（见 §2.5），真机核实前不收紧、也不放宽。
+- `sentinel.py` 是从 P5 分支（25bcd35）原样拿来的超集版本：其中 `capability_for` 的非 one-shot
+  分支在运行时 import `cloud.chv.capabilities`——该模块是 P5 的文件，本轨基线 06c3e17 里没有。
+  本轨只走 one-shot 默认档，运行时不触及该分支（新测试 51/51 证明）；mypy 会报 1 处
+  `import-not-found`，待 P3/P5/P9 集成轨把 capabilities.py 带进来后消除，不在本轨另造一份。
+  其余 mypy 报错与基线一致：`executor.py:1396/1401`（`out` 未注解）与 `marker_tools.py:142`
+  是 P3 已记录的基线问题，本轨未碰。
+
+## 5. 测试清单（`test_chv_muse_p9.py`，全部用标准库与 mock，不联网、不启动真实 VM）
+
+- origin 解析：合法 origin 通过；缺 `SUB2API_BASE_URL` 即错；http 公网主机名/IP 被拒；
+  http 本地中继（字面环回/私网 IP，如 `http://172.17.0.1:28084`）通过且端口保留；
+  URL 带 userinfo 被拒；尾斜杠被剥掉；未知 provider 标签走默认网关（不 deny）。
+- allowlist：`CHV_INFERENCE_ALLOW_HOSTS` 显式覆盖生效；默认即 origin host。
+- 上限校验：超时/重试非法值（非整数、越界）即 ValueError；默认值 120/1。
+- 预审决策（假解析器，不联网）：allow（公网 IP + 白名单 + 路径内）；deny（解析到私网 IP、
+  origin 非法）；ask（主机不在白名单、provider 为空且未配 `CHV_INFERENCE_ORIGIN`）；
+  路径越界（`GET /v1/admin`）即 deny；能力绑定（单次、300 秒）；本地中继 allow（不查 DNS，
+  用必错解析器钉住）与本地中继 ask（allowlist 不含该主机）。
+- 旁路：直连（非通道）调模型无入口——执行器 `_run_with_vm` 前必过预审（mock 执行器验调用顺序：
+  预审 deny 时 `wake_vm` 未被调用）；任务级 `extra_env` 的 origin 被合并进预审（worker 网关
+  路径），非法 origin 转 deny 而不是裸 ValueError。
+- 遥测：白名单外键丢弃；`prompt`/`content`/`api_key` 类键丢弃；采样确定性（同一 task_id 多次判定一致；
+  采样率 0 全丢、1 全留）；落盘失败（state_dir 不可写）不抛错、不影响任务；token 未回传记 `null`。
+- 训练开关：默认开；关后轨迹事件不再产生、任务级遥测不受影响；开关变更持久化 + 审计有记录；
+  状态文件损坏时 fail-closed（当作关处理，记日志）。
+- host.py：`_server_command` 的 origin 结构校验（http 公网被拒、本地中继通过、allowlist 不匹配
+  被拒、合法通过）；`Server.request` 默认超时与重试上限走 env（mock 验默认值 120/1 与 env 覆盖）。
+- 旧测试 harness：5 个文件的 `make_exec` 配了假 origin（`SUB2API_BASE_URL` 与
+  `CHV_INFERENCE_ORIGIN` 均为 `https://xgapi.lol`）+ 假公网解析器——这是新 fail-closed 门
+  要求的测试配置，不是放宽门；门的 deny/ask 行为由本轨新测试钉住。
+
+## 6. 真机待验证清单（排队，测试机空闲后；第三棒 → round-11 → P4/P5/P6 → 本轨）
+
+1. `GET /v1/models` 是否必需：真机跑一轮 turn，确认 codex 是否调它；必需则保留，不必需则收紧为只放 `POST /v1/responses`。
+2. 预审对真实 xgapi 主机的判定：`https://xgapi.lol` 解析后 IP 是否干净（不在 SSRF 挡截网段内）；若解析到被挡网段，记录真实判定、不绕过。
+3. `CHV_INFERENCE_ORIGIN` 未配时 xcmapi 任务是否如实停在预审（ask + 审计），配上后是否放行。
+4. 本地中继真机：`SUB2API_BASE_URL=http://172.17.0.1:28084` 的任务预审是否放行（本地中继分支）、
+   真实 turn 是否走通该中继；同时确认 http 公网 origin 在真机仍被拒。
+5. 真实 turn 的 usage 是否回传：遥测里的 token 字段是否从 `null` 变成实数。
+6. 遥测落盘文件在真实任务后是否存在、内容是否只有白名单字段。
+7. 训练开关在真机上的持久化：开关翻转后重启执行器，状态是否还在。
+7. 超时/重试上限在真机上的表现：长 turn 是否在 `CHV_INFERENCE_TIMEOUT` 处超时（只记录、不主动造超时）。

host.py

@@ -45,6 +45,7 @@ CODEX_BIN=.../bin/codex（start.sh 已自动优先用 bin/codex）。
 """
 import argparse
 import contextlib
+import ipaddress
 import json
 import os
 import queue
@@ -56,6 +57,7 @@ import threading
 import time
 import uuid
 from concurrent.futures import ThreadPoolExecutor
+from urllib.parse import urlparse
 
 HERE = os.path.dirname(os.path.abspath(__file__))
 DEFAULT_STATE = os.path.join(HERE, "host-state.json")
@@ -338,6 +340,81 @@ def _append_trace_marker(state_path, task_name, entry):
         pass
 
 
+def _inference_request_timeout(timeout):
+    """单次请求超时（秒）：显式传参优先；否则读
+    `CHV_INFERENCE_TIMEOUT`（默认 120，与旧默认值一致），
+    非整数或越界（10–600）即 ValueError。"""
+    if timeout is not None:
+        return timeout
+    raw = os.environ.get("CHV_INFERENCE_TIMEOUT", "").strip()
+    if not raw:
+        return 120
+    try:
+        value = int(raw)
+    except ValueError:
+        raise ValueError(
+            "CHV_INFERENCE_TIMEOUT 必须是整数: %r" % raw)
+    if not 10 <= value <= 600:
+        raise ValueError(
+            "CHV_INFERENCE_TIMEOUT 必须在 10-600 之间: %s" % value)
+    return value
+
+
+def _inference_max_retries():
+    """BrokenPipe 重连上限（次）：读 `CHV_INFERENCE_MAX_RETRIES`，
+    默认 1（即现有行为 `range(2)`），非整数或越界（0–5）即
+    ValueError。模型回合级重试仍是 worker 的 max_retries 语义，
+    这里只 bound 传输层重连。"""
+    raw = os.environ.get("CHV_INFERENCE_MAX_RETRIES", "").strip()
+    if not raw:
+        return 1
+    try:
+        value = int(raw)
+    except ValueError:
+        raise ValueError(
+            "CHV_INFERENCE_MAX_RETRIES 必须是整数: %r" % raw)
+    if not 0 <= value <= 5:
+        raise ValueError(
+            "CHV_INFERENCE_MAX_RETRIES 必须在 0-5 之间: %s" % value)
+    return value
+
+
+def _validate_inference_origin(origin):
+    """sub2api origin 结构校验（feat-chv-muse-p9，第二道防线，
+    只用标准库）：https 公网网关（端口只 443）或 http 本地中继
+    （字面环回/私网 IP，如 http://172.17.0.1:28084，见
+    cloud/README.md 的 sub2api 网关一节）；无 userinfo、有主机
+    名；若 `CHV_INFERENCE_ALLOW_HOSTS` 已配，主机必须在其中。
+    第一道防线是执行器的 Sentinel 预审（起 VM 前）；这里拦的
+    是“执行器没审就直接起 codex”的旁路（配置错了 loud fail，
+    不静默跑向别处）。报错不带 origin 原值。"""
+    parsed = urlparse(origin)
+    scheme = parsed.scheme.lower()
+    if "@" in parsed.netloc:
+        raise ValueError("推理 origin 不得带 userinfo（凭证禁进 URL）")
+    host = (parsed.hostname or "").lower()
+    if not host:
+        raise ValueError("推理 origin 无主机名")
+    if scheme == "https":
+        if parsed.port is not None and parsed.port != 443:
+            raise ValueError("推理 origin 端口只接受 443")
+    elif scheme == "http":
+        try:
+            addr = ipaddress.ip_address(host)
+        except ValueError:
+            raise ValueError("http 推理 origin 只接受本地中继"
+                             "（环回/私网 IP）")
+        if not (addr.is_loopback or addr.is_private):
+            raise ValueError("http 推理 origin 只接受本地中继"
+                             "（环回/私网 IP）")
+    else:
+        raise ValueError("推理 origin 必须 https（或本地中继 http）")
+    allowed = os.environ.get("CHV_INFERENCE_ALLOW_HOSTS", "")
+    hosts = {h.strip().lower() for h in allowed.split(",") if h.strip()}
+    if hosts and host not in hosts:
+        raise ValueError("推理 origin 主机不在 allowlist 内: %s" % host)
+
+
 def _server_command(codex_bin, provider=None):
     """app-server 启动命令：-c 必须放在子命令之前。
 
@@ -361,6 +438,9 @@ def _server_command(codex_bin, provider=None):
         if not origin:
             raise RuntimeError(
                 "provider=sub2api 但 SUB2API_BASE_URL 未配置")
+        # feat-chv-muse-p9：origin 结构校验（第二道防线；第一道
+        # 是执行器起 VM 前的 Sentinel 预审）。
+        _validate_inference_origin(origin)
         cmd += [
             "-c", 'model_provider="sub2api"',
             "-c", 'model_providers.sub2api.name="Sub2API"',
@@ -519,14 +599,19 @@ class Server:
             self.proc.stdin.write((json.dumps(obj) + "\n").encode())
             self.proc.stdin.flush()
 
-    def request(self, method, params=None, timeout=120):
+    def request(self, method, params=None, timeout=None):
-        # BrokenPipe / 进程已死时重拉一次再重试（只重试一次，
+        # BrokenPipe / 进程已死时重拉再重试（重试上限见
+        # `_inference_max_retries`，默认 1 次即现有行为；
         # turn/start 这类有副作用的调用由上层 resume 后再跑，
         # 这里重试的是 initialize/list/resume 等幂等请求为主；
         # 即使是 turn/start，重试也只发生在“还没收到任何响应”
         # 的 BrokenPipe 场景，server 死在 turn 中途时 _read_loop
         # 已退出、q.get 会超时而不是走这条路）。
-        for attempt in range(2):
+        # feat-chv-muse-p9：默认超时走 CHV_INFERENCE_TIMEOUT
+        # （默认 120 秒，即旧默认值），显式传参优先。
+        timeout = _inference_request_timeout(timeout)
+        max_retries = _inference_max_retries()
+        for attempt in range(1 + max_retries):
             rid = self.next_id
             self.next_id += 1
             q = queue.Queue()
@@ -539,7 +624,7 @@ class Server:
             except (BrokenPipeError, OSError):
                 with self.pending_lock:
                     self.pending.pop(rid, None)
-                if attempt == 0 and self.ensure_alive():
+                if attempt < max_retries and self.ensure_alive():
                     continue
                 raise
             finally: