fix(telemetry): replace fire-and-forget uvx with bounded queue-based …

.claude-plugin/marketplace.json

@@ -17,7 +17,7 @@
       ],
       "name": "alibabacloud-core",
       "source": "./plugins/alibabacloud-core",
-      "version": "1.0.33"
+      "version": "1.0.34"
     },
     {
       "category": "cloud",

plugins/alibabacloud-core/.claude-plugin/plugin.json

@@ -11,5 +11,5 @@
   "license": "Apache-2.0",
   "name": "alibabacloud-core",
   "repository": "https://github.com/aliyun/alibabacloud-agent-toolkit",
-  "version": "1.0.33"
+  "version": "1.0.34"
 }

plugins/alibabacloud-core/.codex-plugin/plugin.json

@@ -1,6 +1,6 @@
 {
   "name": "alibabacloud-core",
-  "version": "1.0.33",
+  "version": "1.0.34",
   "description": "Core Alibaba Cloud plugin for OpenAPI SDK code generation through a constrained MCP server.",
   "author": {
     "name": "Alibaba Cloud"

plugins/alibabacloud-core/.qoder-plugin/plugin.json

@@ -1,6 +1,6 @@
 {
   "name": "alibabacloud-core",
-  "version": "1.0.33",
+  "version": "1.0.34",
   "description": "Core Alibaba Cloud plugin for OpenAPI SDK code generation through a constrained MCP server.",
   "displayName": "Alibaba Cloud Core",
   "author": {

plugins/alibabacloud-core/hooks/scripts/lib/post_handler.py

@@ -5,7 +5,7 @@
 status, sanitizes outputs, and prints a flat list of CLI args (key on one
 line, value on the next) for the bash wrapper to assemble into:
 
-    uvx alibabacloud.mcp-proxy@latest plugin-telemetry <args>
+    plugin-telemetry <args>  (queued via bounded worker)
 
 Exit codes:
     0 — args printed (caller should upload)

plugins/alibabacloud-core/hooks/scripts/lib/stop_handler.py

@@ -71,14 +71,6 @@ def _iso_from_ms(ms: int) -> str:
     return time.strftime("%Y-%m-%dT%H:%M:%S", t) + f".{millis:03d}Z"
 
 
-def _uploader_cmd() -> list:
-    """Resolve mcp-proxy invocation. Env var lets .sh override for dev."""
-    override = os.environ.get("ALIBABACLOUD_TELEMETRY_UPLOADER")
-    if override:
-        return override.split()
-    return ["uvx", "alibabacloud.mcp-proxy@latest", "plugin-telemetry"]
-
-
 _MCP_SESSION_DIR = os.path.expanduser(
     "~/.cache/alibabacloud-agent-toolkit/mcp-sessions"
 )
@@ -142,38 +134,62 @@ def _strip_optin_fields(args: dict) -> None:
 
 
 def _spawn_upload(args: dict) -> None:
-    """Fire-and-forget mcp-proxy upload for per-call events. The primary
-    user_prompt_turn_start event still flows via stdout to the .sh wrapper —
-    this is only for the N extra llm_call events that don't fit the
-    single-event stdout protocol."""
-    import subprocess
-    argv = list(_uploader_cmd())
+    """Queue an upload event for bounded background processing.
+
+    Replaces the old fire-and-forget uvx invocation that spawned one
+    unbounded process per event, causing orphan process accumulation
+    and disk exhaustion. Events are written to a per-client queue and
+    processed by a single-instance worker with concurrency and timeout
+    controls.
+    """
+    try:
+        from telemetry_enqueue import enqueue_event
+    except ImportError:
+        return
+
+    cdir = _resolve_cdir_for_upload()
+    if not cdir:
+        return
+
+    filtered = {}
     for key in _EMIT_ORDER:
         v = args.get(key)
-        if v is None or v == "":
-            continue
-        argv.append(f"--{key}")
-        argv.append(str(v))
-    log_path = os.environ.get("ALIBABACLOUD_TELEMETRY_UPLOAD_LOG")
-    if log_path:
-        try:
-            out_fd = open(log_path, "ab")
-        except Exception:
-            out_fd = subprocess.DEVNULL
-    else:
-        out_fd = subprocess.DEVNULL
+        if v is not None and v != "":
+            filtered[key] = str(v)
+    if not filtered:
+        return
+
     try:
-        subprocess.Popen(
-            argv,
-            stdin=subprocess.DEVNULL,
-            stdout=out_fd,
-            stderr=out_fd,
-            start_new_session=True,
-        )
+        enqueue_event(cdir, filtered, start_worker=True)
     except Exception:
         pass
 
 
+def _resolve_cdir_for_upload() -> "str | None":
+    """Resolve the per-client state directory for queue writes."""
+    base = os.environ.get("ALIBABACLOUD_TELEMETRY_STATE_DIR")
+    if not base:
+        base = os.path.expanduser(
+            "~/.cache/alibabacloud-agent-toolkit/telemetry"
+        )
+    client = "unknown"
+    if os.environ.get("COPILOT_CLI") == "1":
+        client = "copilot-cli"
+    elif os.environ.get("CODEX_CLI") == "1":
+        client = "codex"
+    elif os.environ.get("QODER_WORK") == "1":
+        client = "qoderwork"
+    else:
+        client = "claude-code"
+    safe = "".join(c if c.isalnum() or c in "_-" else "_" for c in client)[:64]
+    cdir = os.path.join(base, safe)
+    try:
+        os.makedirs(cdir, exist_ok=True)
+    except OSError:
+        return None
+    return cdir
+
+
 def _emit(args: dict) -> None:
     for key in _EMIT_ORDER:
         v = args.get(key)
@@ -343,11 +359,11 @@ def main() -> int:
                     "tool_tokens": {},
                 })
 
-            # --- Remote telemetry: per-LLM-call uploads (fire-and-forget) ---
+            # --- Remote telemetry: per-LLM-call uploads (queue-based) ---
             # These bypass the single-event stdout protocol because the .sh
-            # wrapper only fires one mcp-proxy invocation per hook trigger.
-            # Each call gets its own background uvx process; ordering in SLS
-            # is by start_timestamp (no callIndex needed, no model uploaded).
+            # wrapper only fires one upload per hook trigger. Each call is
+            # queued for the bounded worker; ordering in SLS is by
+            # start_timestamp (no callIndex needed, no model uploaded).
             if turn_has_trace and prompt_span and llm_calls:
                 for call in llm_calls:
                     upload_args = {

plugins/alibabacloud-core/hooks/scripts/lib/telemetry_enqueue.py

@@ -0,0 +1,102 @@
+#!/usr/bin/env python3
+"""Queue a telemetry event for bounded background upload.
+
+Reads --key/value lines from stdin (output of post/prompt/stop handlers)
+and writes them as a JSON file to the per-client queue directory.
+Optionally starts the bounded worker in the background.
+
+Usage:
+    echo -e "--event-type\\nmcp_tool_use\\n--tool-name\\nFoo" | \\
+        python3 telemetry_enqueue.py <cdir> [--no-worker]
+"""
+
+import json
+import os
+import subprocess
+import sys
+import time
+import uuid
+
+
+def enqueue_event(cdir, args_dict, start_worker=True):
+    """Write a single event to the queue directory and optionally start worker."""
+    queue_dir = os.path.join(cdir, "telemetry-queue", "pending")
+    os.makedirs(queue_dir, mode=0o700, exist_ok=True)
+
+    event = {
+        "args": args_dict,
+        "retries": 0,
+        "timestamp": time.time(),
+    }
+
+    filename = f"{int(time.time() * 1000000)}-{uuid.uuid4().hex[:8]}.json"
+    filepath = os.path.join(queue_dir, filename)
+
+    tmp = filepath + ".tmp"
+    with open(tmp, "w") as f:
+        json.dump(event, f)
+    os.rename(tmp, filepath)
+
+    if start_worker:
+        _ensure_worker(cdir)
+
+
+def _ensure_worker(cdir):
+    """Start the bounded worker in the background if not already running."""
+    script_dir = os.path.dirname(os.path.abspath(__file__))
+    worker = os.path.join(script_dir, "telemetry_worker.py")
+
+    env = os.environ.copy()
+    env["ALIBABACLOUD_TELEMETRY_WORKER_STATE_DIR"] = cdir
+    if os.environ.get("ALIBABACLOUD_TELEMETRY_DEBUG") == "1":
+        env["ALIBABACLOUD_TELEMETRY_WORKER_DEBUG"] = "1"
+
+    try:
+        subprocess.Popen(
+            [sys.executable, worker, cdir],
+            stdin=subprocess.DEVNULL,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+            start_new_session=True,
+            env=env,
+            close_fds=True,
+        )
+    except Exception:
+        pass
+
+
+def _parse_lines_to_dict(text):
+    """Parse alternating --key / value lines into a flat dict."""
+    lines = [l for l in text.splitlines() if l]
+    args_dict = {}
+    i = 0
+    while i < len(lines):
+        if lines[i].startswith("--") and i + 1 < len(lines):
+            key = lines[i][2:]
+            args_dict[key] = lines[i + 1]
+            i += 2
+        else:
+            i += 1
+    return args_dict
+
+
+def main():
+    if len(sys.argv) < 2:
+        sys.exit(1)
+
+    cdir = sys.argv[1]
+    no_worker = "--no-worker" in sys.argv
+
+    text = sys.stdin.read()
+    if not text.strip():
+        return
+
+    args_dict = _parse_lines_to_dict(text)
+    if not args_dict:
+        return
+
+    enqueue_event(cdir, args_dict, start_worker=not no_worker)
+
+
+if __name__ == "__main__":
+    main()