refactor(usage): reuse typed device context validation

Signed-off-by: huangruiteng <huangrt01@163.com>

Author: huangruiteng

apps/presentation/dashboard/src/data/chat.ts

@@ -2251,6 +2251,7 @@ export async function disconnectLarkGoalTopic(goalId: string, connectionId: stri
   );
 }
 
+export { CONTEXTS as usageContexts } from "../../../../../loopx/control_plane/runtime/usage_statistics_contract";
 const usageStatisticsSchema = z.object({
   consent: z.enum(["default", "enabled", "disabled"]),
   sending: z.boolean(), blocked_by: z.string().nullable(), endpoint: z.string().nullable(),

apps/presentation/dashboard/src/features/personal-workspace/usage-statistics-settings.tsx

@@ -1,5 +1,5 @@
 import { useEffect, useState } from "react";
-import { setUsageContext, usageStatistics, type UsageStatistics } from "../../data/chat";
+import { setUsageContext, usageContexts, usageStatistics, type UsageStatistics } from "../../data/chat";
 import { useWorkspaceI18n } from "./i18n";
 
 export function UsageStatisticsSettings() {
@@ -36,7 +36,7 @@ export function UsageStatisticsSettings() {
     {state ? <>
       <label>{zh ? "这台设备的用途（自愿声明）" : "Device deployment context (voluntary)"}
         <select value={state.stored_context ?? "unknown"} disabled={busy} onChange={event => void updateContext(event.target.value)}>
-          {(["unknown", "personal", "shared_service", "ephemeral", "organization_managed", "maintainer"] as const).map(value =>
+          {usageContexts.map(value =>
             <option key={value} value={value}>{zh ? ({ unknown: "未知", personal: "个人使用", shared_service: "共享服务",
               ephemeral: "临时环境", organization_managed: "组织管理", maintainer: "项目维护者" })[value] : value}</option>)}
         </select>

loopx/chat_usage_statistics_api.py

@@ -4,7 +4,7 @@
 import subprocess
 from typing import Any
 
-from .usage_ping import control
+from .usage_ping import UsageSettingsInputError, control
 
 CHAT_USAGE_STATISTICS_PATH = "/api/chat/usage-statistics"
 
@@ -24,8 +24,6 @@ def _usage_statistics_update(self) -> None:
         try:
             body = self._read_json()
             if set(body) == {"context"} and isinstance(body["context"], str):
-                if body["context"] not in {"unknown", "personal", "shared_service", "ephemeral", "organization_managed", "maintainer"}:
-                    raise ValueError("invalid deployment context")
                 self._usage_statistics_request("context", context=body["context"])
                 return
             if set(body) == {"notice"} and isinstance(body["notice"], dict):
@@ -41,6 +39,9 @@ def _usage_statistics_update(self) -> None:
     def _usage_statistics_request(self, action: str, **fields: Any) -> None:
         try:
             projection = control(action, **fields)
+        except UsageSettingsInputError as exc:
+            self._send_error(str(exc), status=400, error_code="invalid_usage_settings")
+            return
         except (OSError, ValueError, RuntimeError, TimeoutError, subprocess.TimeoutExpired):
             self._send_error("Usage settings unavailable; use loopx usage-ping status in the terminal.",
                              status=503, error_code="usage_settings_unavailable")

loopx/cli_commands/usage_ping.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import argparse
+import sys
 from collections.abc import Callable
 
 from .. import usage_ping
@@ -49,7 +50,7 @@ def register_usage_ping_command(
         default="status",
         help="status previews payloads; enable accepts collection; disable clears the ID and pending counts.",
     )
-    parser.add_argument("--context", choices=("unknown", "personal", "shared_service", "ephemeral", "organization_managed", "maintainer"),
+    parser.add_argument("--context", metavar="CONTEXT",
                         help="Persistent device label; only with context. Does not enable collection; environment takes precedence.")
     add_subcommand_format(parser)
     return parser
@@ -58,6 +59,10 @@ def register_usage_ping_command(
 def handle_usage_ping_command(args: argparse.Namespace, print_payload: PrintPayload) -> int:
     if (args.action == "context") != (args.context is not None):
         raise ValueError("use usage-ping context --context <value>; other actions take no context")
-    payload = usage_ping.control(args.action, **({"context": args.context} if args.context is not None else {}))
+    try:
+        payload = usage_ping.control(args.action, **({"context": args.context} if args.context is not None else {}))
+    except usage_ping.UsageSettingsInputError as exc:
+        print(str(exc), file=sys.stderr)
+        return 2
     print_payload(payload, output_format(args), render_usage_ping_markdown)
     return 0

loopx/control_plane/runtime/usage_statistics.ts

@@ -139,9 +139,13 @@ export async function inspect(path: string, ctx: Context) {
 export function effectiveContext(state: Pick<State, "context">, ctx: Context): Diagnostic["context"] {
   return usageContext(ctx.env.LOOPX_USAGE_CONTEXT !== undefined ? ctx.env.LOOPX_USAGE_CONTEXT : state.context);
 }
+/** Only this input rejection may cross the adapter as a caller error. */
+export class UsageContextInputError extends Error {
+  constructor() { super("usage_context_invalid"); }
+}
 /** A device label is observation metadata, never consent, identity or work authority. */
 export async function configureContext(path: string, ctx: Context, value: unknown) {
-  if (!(CONTEXTS as readonly unknown[]).includes(value)) throw new Error("usage_context_invalid");
+  if (!(CONTEXTS as readonly unknown[]).includes(value)) throw new UsageContextInputError();
   await withFileMutationLock(path, async () => {
     const state = await load(path);
     state.generation ||= randomUUID();

loopx/control_plane/runtime/usage_statistics_cli.ts

@@ -1,4 +1,4 @@
-import { configure, configureContext, inspect, observe } from "./usage_statistics.ts";
+import { configure, configureContext, inspect, observe, UsageContextInputError } from "./usage_statistics.ts";
 import { profileFeature } from "./usage_statistics_installation_contract.ts";
 import { durationBucket, FEATURES, object } from "./usage_statistics_contract.ts";
 import type { Context } from "./usage_statistics.ts";
@@ -47,8 +47,9 @@ try {
     }
   } else throw new Error("usage_request_invalid");
   process.stdout.write(JSON.stringify(result) + "\n");
-} catch {
+} catch (error) {
   // No environment, source paths or exception text in the diagnostic contract.
-  process.stdout.write(JSON.stringify({ error: "usage_statistics_unavailable" }) + "\n");
+  process.stdout.write(JSON.stringify({ error: error instanceof UsageContextInputError
+    ? "usage_context_invalid" : "usage_statistics_unavailable" }) + "\n");
   process.exitCode = 1;
 }

loopx/control_plane/runtime/usage_statistics_contract.ts

@@ -1,6 +1,7 @@
 /** Public allowlist shared by the client and Cloudflare collector. No text slots. */
 export const PING_SCHEMA = "loopx_usage_ping_v1";
 export const AGGREGATE_SCHEMA = "loopx_usage_aggregate_v1";
+export const CONTEXTS = ["unknown", "personal", "shared_service", "ephemeral", "organization_managed", "maintainer"] as const;
 export const FEATURES = ["status", "quota", "todo", "turn", "project", "connect", "pr-review", "version", "chat", "other"] as const;
 export const OUTCOMES = ["ok", "failed", "cancelled"] as const;
 export const DURATIONS = ["lt_100ms", "lt_1s", "lt_10s", "lt_60s", "gte_60s"] as const;

loopx/control_plane/runtime/usage_statistics_diagnostics.ts

@@ -1,9 +1,9 @@
 /** Versioned, content-free diagnostics. This is observation, never work authority. */
-import { counterKey, DURATIONS, FEATURES, MAX_COUNT, MAX_ROWS, object } from "./usage_statistics_contract.ts";
+import { counterKey, CONTEXTS, DURATIONS, FEATURES, MAX_COUNT, MAX_ROWS, object } from "./usage_statistics_contract.ts";
 import type { Counter } from "./usage_statistics_contract.ts";
 
 export const DIAGNOSTIC_SCHEMA = "loopx_usage_diagnostics_v1";
-export const CONTEXTS = ["unknown", "personal", "shared_service", "ephemeral", "organization_managed", "maintainer"] as const;
+export { CONTEXTS } from "./usage_statistics_contract.ts";
 export const DIAGNOSTIC_FEATURES = [...FEATURES, "heartbeat", "state", "agent", "memory", "capability", "maintenance"] as const;
 export type DiagnosticFeature = typeof DIAGNOSTIC_FEATURES[number];
 /** Fixed parser command names, not argv values or user-supplied extension names. */

loopx/usage_ping.py

@@ -22,6 +22,10 @@
 _observation: ContextVar[dict[str, Any] | None] = ContextVar("usage_observation", default=None)
 
 
+class UsageSettingsInputError(ValueError):
+    """Typed input rejection from the existing TypeScript settings owner."""
+
+
 def select_operation(args: Any) -> None:
     """Parser-owned operation names only; never scan argument values."""
     state = _observation.get()
@@ -105,6 +109,8 @@ def control(action: str, path: Path | None = None, **fields: Any) -> dict[str, A
     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)
+    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.")
     return payload

tests/test_usage_ping.py

@@ -446,6 +446,16 @@ def test_context_setting_real_cli_and_http_share_state_without_enabling(isolated
         assert response.status == 403
         response.read()
         assert usage_ping.control('status')['stored_context'] == 'personal'
+        before = usage_ping.state_path().read_bytes()
+        assert main(['usage-ping', 'context', '--context', 'my-company', '--format', 'json']) == 2
+        assert 'Invalid device deployment context' in capsys.readouterr().err
+        assert usage_ping.state_path().read_bytes() == before
+        usage_ping.state_path().write_text('invalid')
+        connection.request('POST', path, json.dumps({'context': 'personal'}), {'Content-Type': 'application/json'})
+        response = connection.getresponse()
+        assert response.status == 503, 'a damaged store is not a caller input rejection'
+        response.read()
+        assert usage_ping.state_path().read_text() == 'invalid'
     finally:
         connection.close()
         server.shutdown()