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）。
-_CONNECTOR_SCHEMES = {"gitea": frozenset({"https"})}
-_CONNECTOR_PORTS = {"gitea": frozenset({443})}
+# 显式放行时才出现，本期没有任何连接器放 http）。注意：推理的
+# http 本地中继（字面环回/私网 IP，如 http://172.17.0.1:28084）
+# 不走 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