feat(telemetry): integrate telemetry tracking across interactive and …

src/kimi_cli/__main__.py

@@ -10,8 +10,11 @@ def _prog_name() -> str:
 
 
 def main(argv: Sequence[str] | None = None) -> int | str | None:
+    from kimi_cli.telemetry.crash import install_crash_handlers, set_phase
     from kimi_cli.utils.proxy import normalize_proxy_env
 
+    # Install excepthook before anything else so startup-phase crashes are captured.
+    install_crash_handlers()
     normalize_proxy_env()
 
     args = list(sys.argv[1:] if argv is None else argv)
@@ -28,6 +31,8 @@ def main(argv: Sequence[str] | None = None) -> int | str | None:
         return cli(args=args, prog_name=_prog_name())
     except SystemExit as exc:
         return exc.code
+    finally:
+        set_phase("shutdown")
 
 
 if __name__ == "__main__":

src/kimi_cli/acp/server.py

@@ -58,6 +58,14 @@ async def initialize(
         )
         self.client_capabilities = client_capabilities
 
+        if client_info is not None:
+            from kimi_cli.telemetry import set_client_info
+
+            set_client_info(
+                name=client_info.name,
+                version=getattr(client_info, "version", None),
+            )
+
         # get command and args of current process for terminal-auth
         command = sys.argv[0]
         args: list[str] = []
@@ -155,6 +163,7 @@ async def new_session(
         cli_instance = await KimiCLI.create(
             session,
             mcp_configs=[mcp_config],
+            ui_mode="acp",
         )
         config = cli_instance.soul.runtime.config
         acp_kaos = ACPKaos(self.conn, session.id, self.client_capabilities)
@@ -225,6 +234,7 @@ async def _setup_session(
             session,
             mcp_configs=[mcp_config],
             resumed=True,  # _setup_session loads existing sessions
+            ui_mode="acp",
         )
         config = cli_instance.soul.runtime.config
         acp_kaos = ACPKaos(self.conn, session.id, self.client_capabilities)

src/kimi_cli/app.py

@@ -4,6 +4,7 @@
 import contextlib
 import dataclasses
 import sys
+import time
 import warnings
 from collections.abc import AsyncGenerator, Callable
 from pathlib import Path
@@ -14,10 +15,11 @@
 from pydantic import SecretStr
 
 from kimi_cli.agentspec import DEFAULT_AGENT_FILE
-from kimi_cli.auth.oauth import OAuthManager
+from kimi_cli.auth.oauth import KIMI_CODE_OAUTH_KEY, OAuthManager, get_device_id
 from kimi_cli.background.models import is_terminal_status
 from kimi_cli.cli import InputFormat, OutputFormat
 from kimi_cli.config import Config, LLMModel, LLMProvider, load_config
+from kimi_cli.constant import VERSION
 from kimi_cli.llm import augment_provider_with_env_vars, create_llm, model_display_name
 from kimi_cli.session import Session
 from kimi_cli.share import get_share_dir
@@ -26,6 +28,7 @@
 from kimi_cli.soul.context import Context
 from kimi_cli.soul.kimisoul import KimiSoul
 from kimi_cli.utils.aioqueue import QueueShutDown
+from kimi_cli.utils.envvar import get_env_bool
 from kimi_cli.utils.logging import logger, open_original_stderr, redirect_stderr_to_logger
 from kimi_cli.utils.path import shorten_home
 from kimi_cli.wire import Wire, WireUISide
@@ -126,6 +129,7 @@ async def create(
         yolo: bool = False,
         plan_mode: bool = False,
         resumed: bool = False,
+        ui_mode: str = "shell",
         # Extensions
         agent_file: Path | None = None,
         mcp_configs: list[MCPConfig] | list[dict[str, Any]] | None = None,
@@ -174,10 +178,15 @@ async def create(
             MCPRuntimeError(KimiCLIException, RuntimeError): When any MCP server cannot be
                 connected.
         """
+        _create_t0 = time.monotonic()
+        _phase_timings_ms: dict[str, int] = {}
+
         if startup_progress is not None:
             startup_progress("Loading configuration...")
 
+        _phase_t = time.monotonic()
         config = config if isinstance(config, Config) else load_config(config)
+        _phase_timings_ms["config_ms"] = int((time.monotonic() - _phase_t) * 1000)
         if max_steps_per_turn is not None:
             config.loop_control.max_steps_per_turn = max_steps_per_turn
         if max_retries_per_step is not None:
@@ -186,6 +195,7 @@ async def create(
             config.loop_control.max_ralph_iterations = max_ralph_iterations
         logger.info("Loaded config: {config}", config=config)
 
+        _phase_t = time.monotonic()
         oauth = OAuthManager(config)
 
         bg_refresh_task = asyncio.create_task(_refresh_managed_models_silent(config))
@@ -248,6 +258,7 @@ async def create(
         runtime.notifications.recover()
         runtime.background_tasks.reconcile()
         _cleanup_stale_foreground_subagents(runtime)
+        _phase_timings_ms["init_ms"] = int((time.monotonic() - _phase_t) * 1000)
 
         # Refresh plugin configs with fresh credentials (e.g. OAuth tokens)
         try:
@@ -268,12 +279,14 @@ async def create(
         if startup_progress is not None:
             startup_progress("Loading agent...")
 
+        _phase_t = time.monotonic()
         agent = await load_agent(
             agent_file,
             runtime,
             mcp_configs=mcp_configs or [],
             start_mcp_loading=not defer_mcp_loading,
         )
+        _phase_timings_ms["mcp_ms"] = int((time.monotonic() - _phase_t) * 1000)
 
         if startup_progress is not None:
             startup_progress("Restoring conversation...")
@@ -301,6 +314,47 @@ async def create(
         soul.set_hook_engine(hook_engine)
         runtime.hook_engine = hook_engine
 
+        # --- Initialize telemetry ---
+        from kimi_cli.telemetry import attach_sink, set_context
+        from kimi_cli.telemetry import disable as disable_telemetry
+
+        telemetry_disabled = not config.telemetry or get_env_bool("KIMI_DISABLE_TELEMETRY")
+        if telemetry_disabled:
+            disable_telemetry()
+        else:
+            device_id = get_device_id()
+            set_context(device_id=device_id, session_id=session.id)
+            from kimi_cli.telemetry.sink import EventSink
+            from kimi_cli.telemetry.transport import AsyncTransport
+
+            def _get_token() -> str | None:
+                return oauth.get_cached_access_token(KIMI_CODE_OAUTH_KEY)
+
+            transport = AsyncTransport(device_id=device_id, get_access_token=_get_token)
+            sink = EventSink(
+                transport,
+                version=VERSION,
+                model=model.model if model else "",
+                ui_mode=ui_mode,
+            )
+            attach_sink(sink)
+
+        from kimi_cli.telemetry import track
+        from kimi_cli.telemetry.crash import install_asyncio_handler, set_phase
+
+        # App init finished — enter runtime phase and hook asyncio crashes.
+        install_asyncio_handler()
+        set_phase("runtime")
+
+        track("started", resumed=resumed, yolo=yolo)
+        track(
+            "startup_perf",
+            duration_ms=int((time.monotonic() - _create_t0) * 1000),
+            config_ms=_phase_timings_ms.get("config_ms", 0),
+            init_ms=_phase_timings_ms.get("init_ms", 0),
+            mcp_ms=_phase_timings_ms.get("mcp_ms", 0),
+        )
+
         return KimiCLI(soul, runtime, env_overrides, bg_refresh_task)
 
     def __init__(

src/kimi_cli/auth/oauth.py

@@ -209,6 +209,9 @@ def get_device_id() -> str:
     device_id = uuid.uuid4().hex
     path.write_text(device_id, encoding="utf-8")
     _ensure_private_file(path)
+    from kimi_cli.telemetry import track
+
+    track("first_launch")
     return device_id
 
 
@@ -783,6 +786,10 @@ def _cache_access_token(self, ref: OAuthRef, token: OAuthToken) -> None:
             return
         self._access_tokens[ref.key] = token.access_token
 
+    def get_cached_access_token(self, key: str) -> str | None:
+        """Get a cached access token by key, or None if not available."""
+        return self._access_tokens.get(key)
+
     def common_headers(self) -> dict[str, str]:
         return _common_headers()
 
@@ -961,15 +968,24 @@ async def _refresh_tokens(
                         "OAuth credentials rejected: {error}",
                         error=exc,
                     )
+                    from kimi_cli.telemetry import track
+
+                    track("oauth_refresh", success=False, reason="unauthorized")
                     return
                 except Exception as exc:
                     if force:
                         raise
                     logger.warning("Failed to refresh OAuth token: {error}", error=exc)
+                    from kimi_cli.telemetry import track
+
+                    track("oauth_refresh", success=False, reason="network_or_other")
                     return
                 save_tokens(ref, refreshed)
                 self._cache_access_token(ref, refreshed)
                 self._apply_access_token(runtime, refreshed.access_token)
+                from kimi_cli.telemetry import track
+
+                track("oauth_refresh", success=True)
             finally:
                 xlock.release()
 

src/kimi_cli/background/manager.py

@@ -177,6 +177,9 @@ def create_bash_task(
             timeout_s=timeout_s,
         )
         self._store.create_task(spec)
+        from kimi_cli.telemetry import track
+
+        track("background_task_created")
 
         runtime = self._store.read_runtime(task_id)
         task_dir = self._store.task_dir(task_id)
@@ -577,6 +580,11 @@ def _mark_task_completed(self, task_id: str) -> None:
         runtime.finished_at = runtime.updated_at
         runtime.failure_reason = None
         self._store.write_runtime(task_id, runtime)
+        from kimi_cli.telemetry import track
+
+        if runtime.started_at and runtime.finished_at:
+            duration = runtime.finished_at - runtime.started_at
+            track("background_task_completed", success=True, duration_s=duration)
 
     def _mark_task_failed(self, task_id: str, reason: str) -> None:
         runtime = self._store.read_runtime(task_id)
@@ -587,6 +595,16 @@ def _mark_task_failed(self, task_id: str, reason: str) -> None:
         runtime.finished_at = runtime.updated_at
         runtime.failure_reason = reason
         self._store.write_runtime(task_id, runtime)
+        from kimi_cli.telemetry import track
+
+        if runtime.started_at and runtime.finished_at:
+            duration = runtime.finished_at - runtime.started_at
+            track(
+                "background_task_completed",
+                success=False,
+                duration_s=duration,
+                reason="error",
+            )
 
     def _mark_task_timed_out(self, task_id: str, reason: str) -> None:
         runtime = self._store.read_runtime(task_id)
@@ -599,6 +617,16 @@ def _mark_task_timed_out(self, task_id: str, reason: str) -> None:
         runtime.timed_out = True
         runtime.failure_reason = reason
         self._store.write_runtime(task_id, runtime)
+        from kimi_cli.telemetry import track
+
+        if runtime.started_at and runtime.finished_at:
+            duration = runtime.finished_at - runtime.started_at
+            track(
+                "background_task_completed",
+                success=False,
+                duration_s=duration,
+                reason="timeout",
+            )
 
     def _mark_task_killed(self, task_id: str, reason: str) -> None:
         runtime = self._store.read_runtime(task_id)
@@ -610,3 +638,13 @@ def _mark_task_killed(self, task_id: str, reason: str) -> None:
         runtime.interrupted = True
         runtime.failure_reason = reason
         self._store.write_runtime(task_id, runtime)
+        from kimi_cli.telemetry import track
+
+        if runtime.started_at and runtime.finished_at:
+            duration = runtime.finished_at - runtime.started_at
+            track(
+                "background_task_completed",
+                success=False,
+                duration_s=duration,
+                reason="killed",
+            )

src/kimi_cli/cli/__init__.py

@@ -620,6 +620,7 @@ async def _run(session_id: str | None, prefill_text: str | None = None) -> tuple
                 max_ralph_iterations=max_ralph_iterations,
                 startup_progress=startup_progress.update if ui == "shell" else None,
                 defer_mcp_loading=ui == "shell" and prompt is None,
+                ui_mode=ui,
             )
             startup_progress.stop()
 

src/kimi_cli/cli/__main__.py

@@ -5,7 +5,15 @@
 from kimi_cli.cli import cli
 
 if __name__ == "__main__":
+    from kimi_cli.telemetry.crash import install_crash_handlers, set_phase
     from kimi_cli.utils.proxy import normalize_proxy_env
 
+    # Same entry treatment as kimi_cli.__main__: install excepthook before
+    # anything else so startup-phase crashes in subcommand subprocesses
+    # (background-task-worker, __web-worker, acp via toad) are captured.
+    install_crash_handlers()
     normalize_proxy_env()
-    sys.exit(cli())
+    try:
+        sys.exit(cli())
+    finally:
+        set_phase("shutdown")

src/kimi_cli/config.py

@@ -235,6 +235,10 @@ class Config(BaseModel):
             "instead of using only the first one found"
         ),
     )
+    telemetry: bool = Field(
+        default=True,
+        description="Enable anonymous telemetry to help improve kimi-cli. Set to false to disable.",
+    )
 
     @model_validator(mode="after")
     def validate_model(self) -> Self:

src/kimi_cli/hooks/engine.py

@@ -234,13 +234,27 @@ async def trigger(
             return []
 
         try:
-            return await self._execute_hooks(
+            results = await self._execute_hooks(
                 event, matcher_value, server_matched, wire_matched, input_data
             )
         except Exception:
             logger.warning("Hook engine error for {}, failing open", event)
             return []
 
+        # Telemetry runs outside the fail-open try: a telemetry failure
+        # must NEVER discard hook results. For security-critical hooks
+        # (PreToolUse block), treating a sink failure as fail-open would
+        # silently bypass the block.
+        try:
+            from kimi_cli.telemetry import track
+
+            has_block = any(r.action == "block" for r in results)
+            track("hook_triggered", event_type=event, action="block" if has_block else "allow")
+        except Exception:
+            logger.debug("Telemetry for hook_triggered failed")
+
+        return results
+
     async def _execute_hooks(
         self,
         event: str,

src/kimi_cli/session_state.py

@@ -103,8 +103,11 @@ def load_session_state(session_dir: Path) -> SessionState:
         try:
             with open(state_file, encoding="utf-8") as f:
                 state = SessionState.model_validate(json.load(f))
-        except (json.JSONDecodeError, ValidationError, UnicodeDecodeError):
+        except (json.JSONDecodeError, ValidationError, UnicodeDecodeError) as e:
             logger.warning("Corrupted state file, using defaults: {path}", path=state_file)
+            from kimi_cli.telemetry import track
+
+            track("session_load_failed", reason=type(e).__name__)
             state = SessionState()
 
     # One-time migration from legacy metadata.json (best-effort)