fix(usage): return safe settings failures with typed lock recovery

docs/reference/usage-ping.md

@@ -274,6 +274,16 @@ restore withdrawn collection.
 Offline `migrate-local-state` commands do not schedule these observations,
 because their previews and rollback receipts bind the machine-state bytes.
 
+Settings-service and local transport failures exit with status 1. With
+`--format json`, the command returns `ok=false`, `error` and a public-safe
+`message`; Markdown prints the same message. `mutation_lock_timeout` means the
+settings lock was not acquired, so that attempt did not change consent. Wait
+for the current settings operation to finish before retrying. Other unavailable
+results do not confirm a change's outcome: follow the message and inspect
+`loopx usage-ping status` before retrying a change. Child output, raw exceptions
+and local paths are not exposed. Invalid device-context input retains exit
+status 2. Consent, generation fencing and existing lock deadlines are unchanged.
+
 Each installation attempts at most one heartbeat per UTC day. The first measured
 CLI result attempts an aggregate send immediately, including a failed result.
 Later eligible activity sends buffered deltas at most once every 15 minutes;

docs/reference/usage-ping.zh-CN.md

@@ -202,6 +202,14 @@ authority provider 备份、公共投影。普通命令只读取很小的本地
 离线 `migrate-local-state` 命令不调度这些观察，避免后台写入改变预览和回滚收据
 所绑定的整机状态字节。
 
+设置服务或本地传输失败时，命令以状态码 1 退出。`--format json` 返回
+`ok=false`、`error` 和公开安全的 `message`；Markdown 显示同一条消息。
+`mutation_lock_timeout` 表示未取得设置锁，本次尝试没有改变同意状态；等待当前
+设置操作结束后再重试。其他不可用结果不代表已经确认修改结果：按消息恢复，
+重试修改前先用 `loopx usage-ping status` 回读。不暴露子进程输出、原始异常或
+本地路径。无效设备用途输入仍以状态码 2 拒绝；同意策略、generation 隔离及
+既有锁期限保持不变。
+
 每天最多尝试一次心跳。首次可采集的 CLI 命令结束后立即尝试发送汇总，包括失败结果；
 后续有合格活动时，每隔至少 15 分钟发送一批新增计数，UTC 换日不重置间隔。
 心跳和 CLI 汇总分别认领发送机会，心跳尝试不会压掉命令结果。单独启动不产生成功

loopx/cli_commands/usage_ping.py

@@ -16,6 +16,8 @@
 
 
 def render_usage_ping_markdown(payload: dict[str, object]) -> str:
+    if payload.get("ok") is False:
+        return str(payload["message"])
     lines = [f"LoopX basic usage statistics: {payload['consent']}",
              f"- Sending eligible: {payload['sending']}; blocked by: {payload['blocked_by'] or 'none'}",
              f"- Endpoint: {payload['endpoint'] or 'not configured'}",
@@ -64,5 +66,8 @@ def handle_usage_ping_command(args: argparse.Namespace, print_payload: PrintPayl
     except usage_ping.UsageSettingsInputError as exc:
         print(str(exc), file=sys.stderr)
         return 2
+    except usage_ping.UsageSettingsUnavailable as exc:
+        print_payload(exc.as_payload(), output_format(args), render_usage_ping_markdown)
+        return 1
     print_payload(payload, output_format(args), render_usage_ping_markdown)
     return 0

loopx/control_plane/runtime/usage_statistics_cli.ts

@@ -9,6 +9,7 @@ import type { GoalObservation } from "./usage_statistics_goal_contract.ts";
 import { validGoalObservation, hostCategory } from "./usage_statistics_goal_contract.ts";
 
 import { validCycle } from "./usage_statistics_cycles.ts";
+import { EffectRuntimeLockTimeoutError } from "../effect_runtime_errors.ts";
 
 try {
   let input = "";
@@ -50,6 +51,7 @@ try {
 } catch (error) {
   // No environment, source paths or exception text in the diagnostic contract.
   process.stdout.write(JSON.stringify({ error: error instanceof UsageContextInputError
-    ? "usage_context_invalid" : "usage_statistics_unavailable" }) + "\n");
+    ? "usage_context_invalid" : error instanceof EffectRuntimeLockTimeoutError
+      ? error.code : "usage_statistics_unavailable" }) + "\n");
   process.exitCode = 1;
 }

loopx/usage_ping.py

@@ -27,6 +27,22 @@ class UsageSettingsInputError(ValueError):
     """Typed input rejection from the existing TypeScript settings owner."""
 
 
+class UsageSettingsUnavailable(RuntimeError):
+    """Public-safe host failure; never carry child output or exception text."""
+
+    def __init__(self, code: object = None, *, message: str | None = None) -> None:
+        self.code = ("mutation_lock_timeout" if code == "mutation_lock_timeout"
+                     else "usage_statistics_unavailable")
+        super().__init__(message or (
+            "Usage settings are busy. Wait for the current settings operation to finish, then retry."
+            if self.code == "mutation_lock_timeout" else
+            "Usage settings unavailable. Run loopx usage-ping status before retrying a change."
+        ))
+
+    def as_payload(self) -> dict[str, object]:
+        return {"ok": False, "error": self.code, "message": str(self)}
+
+
 def select_operation(args: Any) -> None:
     """Parser-owned operation names only; never scan argument values."""
     state = _observation.get()
@@ -100,20 +116,26 @@ def _request(action: str, path: Path, **fields: Any) -> dict[str, Any]:
 def _command() -> list[str]:
     node = shutil.which("node")
     if node is None:
-        raise RuntimeError("Usage settings require the supported Node.js runtime.")
+        raise UsageSettingsUnavailable(message="Usage settings require the supported Node.js runtime.")
     # Keep the Python sender's proxy behavior when moving I/O to Node. This
     # native flag is available in the repository's supported Node versions.
     return [node, "--no-warnings", "--use-env-proxy", "--experimental-strip-types", str(_ENTRY)]
 
 
 def control(action: str, path: Path | None = None, **fields: Any) -> dict[str, Any]:
-    result = subprocess.run(_command(), input=json.dumps(_request(action, path or state_path(), **fields)),
-                            capture_output=True, text=True, encoding="utf-8", timeout=4, check=False)
-    payload = json.loads(result.stdout)
+    try:
+        result = subprocess.run(_command(), input=json.dumps(_request(action, path or state_path(), **fields)),
+                                capture_output=True, text=True, encoding="utf-8", timeout=4, check=False)
+    except (OSError, subprocess.TimeoutExpired, UnicodeError):
+        raise UsageSettingsUnavailable() from None
+    try:
+        payload = json.loads(result.stdout)
+    except (json.JSONDecodeError, TypeError):
+        raise UsageSettingsUnavailable() from None
     if isinstance(payload, dict) and payload.get("error") == "usage_context_invalid":
         raise UsageSettingsInputError("Invalid device deployment context; see the usage-ping reference.")
     if result.returncode or not isinstance(payload, dict) or "error" in payload:
-        raise RuntimeError("Usage settings unavailable. Inspect the local usage-ping.json; disable can repair invalid state.")
+        raise UsageSettingsUnavailable(payload.get("error") if isinstance(payload, dict) else None)
     return payload
 
 

tests/test_usage_ping.py

@@ -118,6 +118,124 @@ def test_settings_commands_and_corrupt_state_repair(isolated, capsys):
     assert json.loads(capsys.readouterr().out)['next_payload'] is None
 
 
+def test_settings_lock_failure_is_readable_and_does_not_change_consent(isolated, capsys):
+    path = usage_ping.state_path()
+    usage_ping.control('enable')
+    before = path.read_bytes()
+    lock_owner = Path(usage_ping.__file__).parent / 'control_plane/effect_runtime_io.ts'
+    script = '''
+      const {acquireFileMutationLock, releaseFileMutationLock} = await import(process.argv[1]);
+      const lock = await acquireFileMutationLock(process.argv[2]);
+      process.stdout.write('ready\\n');
+      for await (const chunk of process.stdin) { /* Keep the live claim until released. */ }
+      await releaseFileMutationLock(lock.targetPath, lock.token);
+    '''
+    holder = subprocess.Popen(
+        [usage_ping._command()[0], '--no-warnings', '--experimental-strip-types',
+         '--input-type=module', '-e', script, lock_owner.as_uri(), str(path)],
+        stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
+    )
+    ready, line = threading.Event(), []
+
+    def read_ready():
+        line.append(holder.stdout.readline())
+        ready.set()
+
+    threading.Thread(target=read_ready, daemon=True).start()
+    try:
+        assert ready.wait(10) and line == ['ready\n']
+        assert main(['usage-ping', 'disable', '--format', 'json']) == 1
+        captured = capsys.readouterr()
+        failure = json.loads(captured.out)
+        assert failure['ok'] is False and failure['error'] == 'mutation_lock_timeout'
+        assert captured.err == ''
+        assert main(['usage-ping', 'disable', '--format', 'markdown']) == 1
+        assert capsys.readouterr().out.strip() == failure['message']
+        assert path.read_bytes() == before
+        assert usage_ping.control('status')['consent'] == 'enabled'
+    finally:
+        try:
+            holder.communicate(timeout=5)
+        except subprocess.TimeoutExpired:
+            holder.kill()
+            holder.communicate()
+    assert holder.returncode == 0
+    assert main(['usage-ping', 'disable', '--format', 'json']) == 0
+    assert json.loads(capsys.readouterr().out)['consent'] == 'disabled'
+    assert 'install_id' not in json.loads(path.read_text())
+
+
+def test_unavailable_settings_are_public_safe_and_explicit_disable_still_repairs(isolated, capsys):
+    private_content = 'synthetic-private-setting-token'
+    usage_ping.state_path().write_text(private_content)
+    for output_format in ('json', 'markdown'):
+        assert main(['usage-ping', 'status', '--format', output_format]) == 1
+        captured = capsys.readouterr()
+        assert private_content not in captured.out + captured.err
+        assert str(isolated) not in captured.out + captured.err
+        assert 'Traceback' not in captured.out + captured.err
+        if output_format == 'json':
+            failure = json.loads(captured.out)
+            assert failure['ok'] is False and failure['error'] == 'usage_statistics_unavailable'
+        else:
+            assert captured.out.strip() == failure['message']
+        assert captured.err == ''
+    assert usage_ping.state_path().read_text() == private_content
+    assert main(['usage-ping', 'disable', '--format', 'json']) == 0
+    assert json.loads(capsys.readouterr().out)['consent'] == 'disabled'
+
+
+@pytest.mark.parametrize('failure', ['bad_json', 'bad_utf8', 'missing_node'])
+def test_settings_bridge_failure_has_no_traceback_or_private_output(isolated, monkeypatch, capsys, failure):
+    secret = 'synthetic-private-transport-token'
+    if failure == 'missing_node':
+        monkeypatch.setattr(usage_ping.shutil, 'which', lambda *_args, **_kwargs: None)
+    else:
+        body = (f"import sys;sys.stdout.buffer.write(b'\\xff{secret}')"
+                if failure == 'bad_utf8' else f"import sys;print({secret!r});print({secret!r},file=sys.stderr)")
+        monkeypatch.setattr(usage_ping, '_command', lambda: [sys.executable, '-c', body])
+    assert main(['usage-ping', 'status', '--format', 'json']) == 1
+    captured = capsys.readouterr()
+    failure_payload = json.loads(captured.out)
+    assert failure_payload['ok'] is False
+    assert failure_payload['error'] == 'usage_statistics_unavailable'
+    assert captured.err == '' and secret not in captured.out
+    assert str(isolated) not in captured.out and 'Traceback' not in captured.out
+    assert not usage_ping.state_path().exists()
+
+
+def test_lost_settings_reply_preserves_the_choice_and_requires_readback(isolated, monkeypatch, capsys):
+    usage_ping.control('enable')
+    node = usage_ping._command()
+    attempts = isolated / 'settings-attempts.txt'
+    # Run the real TS mutation, then lose its reply within the existing 4s
+    # bridge deadline. The caller cannot infer whether that mutation committed.
+    delayed_reply = '''
+import json, subprocess, sys, time
+from pathlib import Path
+result = subprocess.run(json.loads(sys.argv[1]), input=sys.stdin.read(), capture_output=True, text=True)
+with Path(sys.argv[2]).open('a') as output:
+    output.write(str(result.returncode) + '\\n')
+time.sleep(10)
+'''
+    with monkeypatch.context() as patch:
+        patch.setattr(usage_ping, '_command', lambda: [
+            sys.executable, '-c', delayed_reply, json.dumps(node), str(attempts),
+        ])
+        assert main(['usage-ping', 'disable', '--format', 'json']) == 1
+    captured = capsys.readouterr()
+    failure = json.loads(captured.out)
+    assert failure['ok'] is False and failure['error'] == 'usage_statistics_unavailable'
+    assert 'consent' not in failure and captured.err == ''
+    assert attempts.read_text() == '0\n'  # One successful backend call, no retry.
+    after = usage_ping.state_path().read_bytes()
+    assert main(['usage-ping', 'status', '--format', 'json']) == 0
+    status = json.loads(capsys.readouterr().out)
+    assert status['consent'] == 'disabled' and status['sending'] is False
+    assert status['next_payload'] is None
+    assert usage_ping.state_path().read_bytes() == after
+
+
 @pytest.mark.parametrize('interactive', [False, True])
 def test_first_cli_command_discloses_before_automatic_enable(isolated, monkeypatch, capsys, interactive):
     monkeypatch.setattr(sys.stderr, 'isatty', lambda: interactive)