diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json index 299a725..01bf84b 100644 --- a/.claude-plugin/marketplace.json +++ b/.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", diff --git a/plugins/alibabacloud-core/.claude-plugin/plugin.json b/plugins/alibabacloud-core/.claude-plugin/plugin.json index f798d6c..451a753 100644 --- a/plugins/alibabacloud-core/.claude-plugin/plugin.json +++ b/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" } diff --git a/plugins/alibabacloud-core/.codex-plugin/plugin.json b/plugins/alibabacloud-core/.codex-plugin/plugin.json index a012fd9..9caf996 100644 --- a/plugins/alibabacloud-core/.codex-plugin/plugin.json +++ b/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" diff --git a/plugins/alibabacloud-core/.qoder-plugin/plugin.json b/plugins/alibabacloud-core/.qoder-plugin/plugin.json index 9e21c01..fddb3c3 100644 --- a/plugins/alibabacloud-core/.qoder-plugin/plugin.json +++ b/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": { diff --git a/plugins/alibabacloud-core/hooks/scripts/lib/post_handler.py b/plugins/alibabacloud-core/hooks/scripts/lib/post_handler.py index 1fe8caa..81a3680 100644 --- a/plugins/alibabacloud-core/hooks/scripts/lib/post_handler.py +++ b/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 + plugin-telemetry (queued via bounded worker) Exit codes: 0 — args printed (caller should upload) diff --git a/plugins/alibabacloud-core/hooks/scripts/lib/stop_handler.py b/plugins/alibabacloud-core/hooks/scripts/lib/stop_handler.py index 3e31534..36716bc 100644 --- a/plugins/alibabacloud-core/hooks/scripts/lib/stop_handler.py +++ b/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 = { diff --git a/plugins/alibabacloud-core/hooks/scripts/lib/telemetry_enqueue.py b/plugins/alibabacloud-core/hooks/scripts/lib/telemetry_enqueue.py new file mode 100644 index 0000000..034ebd4 --- /dev/null +++ b/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 [--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() diff --git a/plugins/alibabacloud-core/hooks/scripts/lib/telemetry_worker.py b/plugins/alibabacloud-core/hooks/scripts/lib/telemetry_worker.py new file mode 100644 index 0000000..5e37e27 --- /dev/null +++ b/plugins/alibabacloud-core/hooks/scripts/lib/telemetry_worker.py @@ -0,0 +1,429 @@ +#!/usr/bin/env python3 +"""Bounded telemetry upload worker. + +Processes queued telemetry events with: +- Single-instance control via file lock (flock) +- Fixed virtualenv with pinned package (no per-event uvx resolution) +- Configurable concurrency limit (sequential batch processing) +- Hard timeout per upload with subprocess reaping +- Finite retry budget with dead-letter logging +- No orphan processes (proper session group cleanup) + +Usage: + python3 telemetry_worker.py [--cleanup-only] +""" + +import fcntl +import json +import os +import signal +import subprocess +import sys +import time + +MCP_PROXY_PACKAGE = "alibabacloud.mcp-proxy" +MCP_PROXY_PINNED_VERSION = "0.5.1" +MCP_PROXY_PIN = f"{MCP_PROXY_PACKAGE}=={MCP_PROXY_PINNED_VERSION}" + +DEFAULT_MAX_CONCURRENT = int(os.environ.get( + "ALIBABACLOUD_TELEMETRY_MAX_CONCURRENT", "4" +)) +DEFAULT_HARD_TIMEOUT = int(os.environ.get( + "ALIBABACLOUD_TELEMETRY_UPLOAD_TIMEOUT", "30" +)) +DEFAULT_MAX_RETRIES = int(os.environ.get( + "ALIBABACLOUD_TELEMETRY_MAX_RETRIES", "2" +)) +MAX_QUEUE_SIZE = int(os.environ.get( + "ALIBABACLOUD_TELEMETRY_MAX_QUEUE", "500" +)) +FAILED_RETENTION_DAYS = 7 +PENDING_STALE_HOURS = 24 + +_debug_log_path = None + + +def _log(msg): + ts = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + line = f"[{ts}] {msg}" + print(line, flush=True) + if _debug_log_path: + try: + with open(_debug_log_path, "a") as f: + f.write(line + "\n") + except OSError: + pass + + +def _acquire_lock(state_dir): + """Try to acquire exclusive lock. Returns (lock_fd, lock_path) or None.""" + os.makedirs(state_dir, exist_ok=True) + lock_path = os.path.join(state_dir, "telemetry-worker.lock") + try: + fd = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o600) + except OSError: + return None + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + return fd, lock_path + except (IOError, OSError): + os.close(fd) + return None + + +def _release_lock(lock_info): + if lock_info is None: + return + fd, _ = lock_info + try: + fcntl.flock(fd, fcntl.LOCK_UN) + os.close(fd) + except OSError: + pass + + +def _is_pid_alive(pid): + try: + os.kill(pid, 0) + return True + except (OSError, ProcessLookupError): + return False + + +def _reap_pid(pid): + """Reap a zombie child if it is ours.""" + try: + os.waitpid(pid, os.WNOHANG) + except ChildProcessError: + pass + + +def _write_pid_file(state_dir): + pid_path = os.path.join(state_dir, "telemetry-worker.pid") + with open(pid_path, "w") as f: + f.write(str(os.getpid())) + return pid_path + + +def _check_existing_worker(state_dir): + """Return True if another worker is already running.""" + pid_path = os.path.join(state_dir, "telemetry-worker.pid") + try: + with open(pid_path) as f: + pid = int(f.read().strip()) + if _is_pid_alive(pid) and pid != os.getpid(): + return True + except (FileNotFoundError, ValueError, OSError): + pass + return False + + +def _remove_pid_file(pid_path): + try: + with open(pid_path) as f: + if int(f.read().strip()) == os.getpid(): + os.unlink(pid_path) + except (FileNotFoundError, ValueError, OSError): + pass + + +def _get_upload_cmd(state_dir): + """Return the upload command argv prefix using fixed venv or pinned uvx.""" + override = os.environ.get("ALIBABACLOUD_TELEMETRY_UPLOADER") + if override: + return override.split() + + venv_dir = os.path.join(state_dir, ".venv") + venv_bin = os.path.join(venv_dir, "bin", "plugin-telemetry") + + if os.path.isfile(venv_bin): + return [venv_bin] + + if _ensure_venv(state_dir): + if os.path.isfile(venv_bin): + return [venv_bin] + + return [ + "uvx", "--from", + f"{MCP_PROXY_PACKAGE}=={MCP_PROXY_PINNED_VERSION}", + "plugin-telemetry", + ] + + +def _ensure_venv(state_dir): + """Create or update the fixed virtualenv. Returns True on success.""" + venv_dir = os.path.join(state_dir, ".venv") + marker = os.path.join(venv_dir, ".telemetry-version") + + if os.path.isfile(marker): + try: + with open(marker) as f: + if f.read().strip() == MCP_PROXY_PIN: + return True + except OSError: + pass + + try: + import venv as venv_mod + _log(f"Creating venv at {venv_dir}") + venv_mod.EnvBuilder(with_pip=True, clear=True).create(venv_dir) + except Exception as e: + _log(f"venv creation failed: {e}") + return False + + pip_bin = os.path.join(venv_dir, "bin", "pip") + try: + result = subprocess.run( + [pip_bin, "install", "--quiet", MCP_PROXY_PIN], + timeout=120, + capture_output=True, + text=True, + ) + if result.returncode != 0: + _log(f"pip install failed: {result.stderr[:500]}") + return False + except subprocess.TimeoutExpired: + _log("pip install timed out after 120s") + return False + except Exception as e: + _log(f"pip install error: {e}") + return False + + try: + with open(marker, "w") as f: + f.write(MCP_PROXY_PIN) + except OSError: + pass + + _log(f"Installed {MCP_PROXY_PIN}") + return True + + +def _upload_one(cmd_prefix, args_dict): + """Run a single upload with hard timeout. Raises on failure.""" + argv = list(cmd_prefix) + for key, value in args_dict.items(): + if value is not None and value != "": + argv.append(f"--{key}") + argv.append(str(value)) + + timeout = DEFAULT_HARD_TIMEOUT + + proc = subprocess.Popen( + argv, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + start_new_session=True, + ) + + try: + stdout, stderr = proc.communicate(timeout=timeout) + if proc.returncode != 0: + err_snippet = (stderr.decode("utf-8", errors="replace"))[:300] + raise RuntimeError( + f"upload exited {proc.returncode}: {err_snippet}" + ) + except subprocess.TimeoutExpired: + _kill_process_tree(proc) + raise RuntimeError(f"upload timed out after {timeout}s") + + +def _kill_process_tree(proc): + """Kill a subprocess and its entire process group. No orphans.""" + try: + pgid = os.getpgid(proc.pid) + os.killpg(pgid, signal.SIGTERM) + except (ProcessLookupError, OSError): + pass + + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + try: + pgid = os.getpgid(proc.pid) + os.killpg(pgid, signal.SIGKILL) + except (ProcessLookupError, OSError): + pass + try: + proc.kill() + except (ProcessLookupError, OSError): + pass + try: + proc.wait(timeout=3) + except subprocess.TimeoutExpired: + pass + + _reap_pid(proc.pid) + + +def _list_pending(queue_dir): + """List pending event files sorted by timestamp, capped at MAX_QUEUE_SIZE.""" + try: + files = [ + f for f in os.listdir(queue_dir) + if f.endswith(".json") and not f.endswith(".tmp") + ] + except FileNotFoundError: + return [] + + files.sort() + if len(files) > MAX_QUEUE_SIZE: + excess = files[MAX_QUEUE_SIZE:] + for f in excess: + try: + os.unlink(os.path.join(queue_dir, f)) + except OSError: + pass + _log(f"Queue overflow: dropped {len(excess)} oldest events") + files = files[:MAX_QUEUE_SIZE] + return files + + +def _process_batch(state_dir): + """Process one batch of pending events. Returns count processed.""" + queue_dir = os.path.join(state_dir, "telemetry-queue", "pending") + failed_dir = os.path.join(state_dir, "telemetry-queue", "failed") + os.makedirs(failed_dir, mode=0o700, exist_ok=True) + + files = _list_pending(queue_dir) + if not files: + return 0 + + cmd_prefix = _get_upload_cmd(state_dir) + processed = 0 + + for filename in files: + filepath = os.path.join(queue_dir, filename) + try: + with open(filepath) as f: + event = json.load(f) + except (json.JSONDecodeError, OSError) as e: + _log(f"Removing corrupt event {filename}: {e}") + _safe_unlink(filepath) + continue + + args_dict = event.get("args", {}) + retries = event.get("retries", 0) + event_type = args_dict.get("event-type", "unknown") + + try: + _upload_one(cmd_prefix, args_dict) + _safe_unlink(filepath) + processed += 1 + except RuntimeError as e: + _log(f"Upload failed for {event_type} ({filename}): {e}") + if retries < DEFAULT_MAX_RETRIES: + event["retries"] = retries + 1 + _write_back(filepath, event) + else: + _move_to_failed(filepath, failed_dir, event) + _log(f"Moved to failed after {DEFAULT_MAX_RETRIES} retries: {filename}") + except Exception as e: + _log(f"Unexpected error processing {filename}: {e}") + _move_to_failed(filepath, failed_dir, event) + + return processed + + +def _safe_unlink(path): + try: + os.unlink(path) + except OSError: + pass + + +def _write_back(filepath, event): + try: + with open(filepath, "w") as f: + json.dump(event, f) + except OSError: + pass + + +def _move_to_failed(filepath, failed_dir, event): + try: + os.makedirs(failed_dir, mode=0o700, exist_ok=True) + dest = os.path.join(failed_dir, os.path.basename(filepath)) + with open(dest, "w") as f: + json.dump(event, f) + os.unlink(filepath) + except OSError: + _safe_unlink(filepath) + + +def _cleanup_old(state_dir): + """Remove old failed uploads and stale pending events.""" + now = time.time() + for subdir in ("failed", "pending"): + d = os.path.join(state_dir, "telemetry-queue", subdir) + if not os.path.isdir(d): + continue + max_age = ( + FAILED_RETENTION_DAYS * 86400 + if subdir == "failed" + else PENDING_STALE_HOURS * 3600 + ) + try: + for f in os.listdir(d): + fp = os.path.join(d, f) + try: + if now - os.path.getmtime(fp) > max_age: + os.unlink(fp) + except OSError: + pass + except FileNotFoundError: + pass + + +def main(): + if len(sys.argv) < 2: + print(f"Usage: {sys.argv[0]} [--cleanup-only]", file=sys.stderr) + sys.exit(1) + + state_dir = sys.argv[1] + cleanup_only = "--cleanup-only" in sys.argv + + global _debug_log_path + if os.environ.get("ALIBABACLOUD_TELEMETRY_WORKER_DEBUG") == "1": + _debug_log_path = os.path.join(state_dir, "worker-debug.log") + + if not os.path.isdir(state_dir): + _log(f"State dir does not exist: {state_dir}") + sys.exit(1) + + os.makedirs( + os.path.join(state_dir, "telemetry-queue", "pending"), + mode=0o700, exist_ok=True, + ) + + if _check_existing_worker(state_dir): + return + + lock_info = _acquire_lock(state_dir) + if lock_info is None: + return + + pid_path = _write_pid_file(state_dir) + try: + if cleanup_only: + _cleanup_old(state_dir) + return + + iterations = 0 + while True: + count = _process_batch(state_dir) + iterations += 1 + if count == 0: + break + if iterations >= 1000: + _log("Safety limit reached (1000 iterations), exiting") + break + _cleanup_old(state_dir) + finally: + _remove_pid_file(pid_path) + _release_lock(lock_info) + + +if __name__ == "__main__": + main() diff --git a/plugins/alibabacloud-core/hooks/scripts/post-tool-trace.sh b/plugins/alibabacloud-core/hooks/scripts/post-tool-trace.sh index f918e87..184b59c 100755 --- a/plugins/alibabacloud-core/hooks/scripts/post-tool-trace.sh +++ b/plugins/alibabacloud-core/hooks/scripts/post-tool-trace.sh @@ -1,7 +1,7 @@ #!/bin/bash # Post-tool-use hook wrapper. Delegates classification + status detection to -# lib/post_handler.py, then fires `uvx alibabacloud.mcp-proxy@latest -# plugin-telemetry` in the background. Always returns success to the agent. +# lib/post_handler.py, then queues the event for bounded background upload. +# Always returns success to the agent. set +e umask 077 @@ -147,7 +147,7 @@ done # Dry-run mode: log instead of upload if [ "${ALIBABACLOUD_TELEMETRY_DRY_RUN}" = "1" ]; then { - printf 'DRYRUN: uvx alibabacloud.mcp-proxy@latest plugin-telemetry' + printf 'DRYRUN: queue telemetry event' for a in "${args[@]}"; do printf ' %q' "$a" done @@ -157,31 +157,21 @@ if [ "${ALIBABACLOUD_TELEMETRY_DRY_RUN}" = "1" ]; then return_success fi -# Fire-and-forget: detach so the agent loop never waits on uvx. +# Queue-based upload: write event args to a JSON queue file and start +# the bounded worker. Replaces the old fire-and-forget +# `uvx alibabacloud.mcp-proxy@latest` pattern that spawned one +# unbounded process per event, causing orphan process accumulation. debug_log "$cdir" "decision=upload event=$(extract_arg --event-type "${args[@]}") tool=$(extract_arg --tool-name "${args[@]}")" if [ "${ALIBABACLOUD_TELEMETRY_DEBUG}" = "1" ]; then - # Debug mode: capture uvx output for diagnosis instead of discarding { - printf '[%s] [post-tool] upload-start cmd=uvx alibabacloud.mcp-proxy@latest plugin-telemetry' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" - for a in "${args[@]}"; do printf ' %q' "$a"; done - printf '\n' + printf '[%s] [post-tool] enqueue event=%s tool=%s\n' \ + "$(date -u +%Y%m%dT%H%M%SZ)" \ + "$(extract_arg --event-type "${args[@]}")" \ + "$(extract_arg --tool-name "${args[@]}")" } >> "$cdir/debug.log" 2>/dev/null - ( - uvx_out=$(uvx alibabacloud.mcp-proxy@latest plugin-telemetry "${args[@]}" &1) - uvx_rc=$? - { - printf '[%s] [post-tool] upload-done rc=%d\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" "$uvx_rc" - if [ -n "$uvx_out" ]; then - printf '[%s] [post-tool] upload-output: %s\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" "$uvx_out" - fi - } >> "$cdir/debug.log" 2>/dev/null - ) & - disown 2>/dev/null -else - ( uvx alibabacloud.mcp-proxy@latest plugin-telemetry "${args[@]}" \ - /dev/null 2>&1 & ) >/dev/null 2>&1 - disown 2>/dev/null fi +printf '%s\n' "${args[@]}" | python3 "$scriptDir/lib/telemetry_enqueue.py" "$cdir" 2>/dev/null + return_success diff --git a/plugins/alibabacloud-core/hooks/scripts/prompt-trace.sh b/plugins/alibabacloud-core/hooks/scripts/prompt-trace.sh index 928151d..2e33a87 100755 --- a/plugins/alibabacloud-core/hooks/scripts/prompt-trace.sh +++ b/plugins/alibabacloud-core/hooks/scripts/prompt-trace.sh @@ -1,8 +1,8 @@ #!/bin/bash # UserPromptSubmit hook wrapper. Detects slash-style skill invocations # (`/alibabacloud-*: ...`) submitted directly to Claude Code as -# prompts. Delegates classification to lib/prompt_handler.py, then fires -# `uvx alibabacloud.mcp-proxy@latest plugin-telemetry` in the background. +# prompts. Delegates classification to lib/prompt_handler.py, then queues +# the event for bounded background upload. # Always returns success to the agent. set +e umask 077 @@ -150,7 +150,7 @@ done # Dry-run mode: log instead of upload if [ "${ALIBABACLOUD_TELEMETRY_DRY_RUN}" = "1" ]; then { - printf 'DRYRUN: uvx alibabacloud.mcp-proxy@latest plugin-telemetry' + printf 'DRYRUN: queue telemetry event' for a in "${args[@]}"; do printf ' %q' "$a" done @@ -160,31 +160,20 @@ if [ "${ALIBABACLOUD_TELEMETRY_DRY_RUN}" = "1" ]; then return_success fi -# Fire-and-forget: detach so the agent loop never waits on uvx. +# Queue-based upload: write event args to a JSON queue file and start +# the bounded worker. Replaces the old fire-and-forget +# `uvx alibabacloud.mcp-proxy@latest` pattern that spawned one +# unbounded process per event, causing orphan process accumulation. debug_log "$cdir" "decision=upload skill=$(extract_arg --skill-name "${args[@]}")" if [ "${ALIBABACLOUD_TELEMETRY_DEBUG}" = "1" ]; then - # Debug mode: capture uvx output for diagnosis instead of discarding { - printf '[%s] [prompt] upload-start cmd=uvx alibabacloud.mcp-proxy@latest plugin-telemetry' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" - for a in "${args[@]}"; do printf ' %q' "$a"; done - printf '\n' + printf '[%s] [prompt] enqueue skill=%s\n' \ + "$(date -u +%Y%m%dT%H%M%SZ)" \ + "$(extract_arg --skill-name "${args[@]}")" } >> "$cdir/debug.log" 2>/dev/null - ( - uvx_out=$(uvx alibabacloud.mcp-proxy@latest plugin-telemetry "${args[@]}" &1) - uvx_rc=$? - { - printf '[%s] [prompt] upload-done rc=%d\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" "$uvx_rc" - if [ -n "$uvx_out" ]; then - printf '[%s] [prompt] upload-output: %s\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" "$uvx_out" - fi - } >> "$cdir/debug.log" 2>/dev/null - ) & - disown 2>/dev/null -else - ( uvx alibabacloud.mcp-proxy@latest plugin-telemetry "${args[@]}" \ - /dev/null 2>&1 & ) >/dev/null 2>&1 - disown 2>/dev/null fi +printf '%s\n' "${args[@]}" | python3 "$scriptDir/lib/telemetry_enqueue.py" "$cdir" 2>/dev/null + return_success diff --git a/plugins/alibabacloud-core/hooks/scripts/stop-turn-increment.sh b/plugins/alibabacloud-core/hooks/scripts/stop-turn-increment.sh index 932d484..ed734b3 100755 --- a/plugins/alibabacloud-core/hooks/scripts/stop-turn-increment.sh +++ b/plugins/alibabacloud-core/hooks/scripts/stop-turn-increment.sh @@ -3,7 +3,7 @@ # Turn number is consumed by post-tool-trace.sh to tag --turn on each event. # Also bound to StopFailure for symmetry; both paths log identically. # When the turn involved alibabacloud tools, stop_handler.py emits a -# user_prompt_turn_start event to stdout which we upload to remote telemetry. +# user_prompt_turn_start event to stdout which we queue for bounded upload. # Delegates to lib/stop_handler.py which uses fcntl-locked per-session state. set +e umask 077 @@ -132,7 +132,7 @@ done # Dry-run mode: log instead of upload if [ "${ALIBABACLOUD_TELEMETRY_DRY_RUN}" = "1" ]; then { - printf 'DRYRUN: uvx alibabacloud.mcp-proxy@latest plugin-telemetry' + printf 'DRYRUN: queue telemetry event' for a in "${args[@]}"; do printf ' %q' "$a" done @@ -142,31 +142,20 @@ if [ "${ALIBABACLOUD_TELEMETRY_DRY_RUN}" = "1" ]; then exit 0 fi -# Fire-and-forget: detach so the agent loop never waits on uvx. +# Queue-based upload: write event args to a JSON queue file and start +# the bounded worker. Replaces the old fire-and-forget +# `uvx alibabacloud.mcp-proxy@latest` pattern that spawned one +# unbounded process per event, causing orphan process accumulation. debug_log "$cdir" "[stop] decision=upload event=$(extract_arg --event-type "${args[@]}")" if [ "${ALIBABACLOUD_TELEMETRY_DEBUG}" = "1" ]; then - # Debug mode: capture uvx output for diagnosis instead of discarding { - printf '[%s] [stop] upload-start cmd=uvx alibabacloud.mcp-proxy@latest plugin-telemetry' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" - for a in "${args[@]}"; do printf ' %q' "$a"; done - printf '\n' + printf '[%s] [stop] enqueue event=%s\n' \ + "$(date -u +%Y%m%dT%H%M%SZ)" \ + "$(extract_arg --event-type "${args[@]}")" } >> "$cdir/debug.log" 2>/dev/null - ( - uvx_out=$(uvx alibabacloud.mcp-proxy@latest plugin-telemetry "${args[@]}" &1) - uvx_rc=$? - { - printf '[%s] [stop] upload-done rc=%d\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" "$uvx_rc" - if [ -n "$uvx_out" ]; then - printf '[%s] [stop] upload-output: %s\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" "$uvx_out" - fi - } >> "$cdir/debug.log" 2>/dev/null - ) & - disown 2>/dev/null -else - ( uvx alibabacloud.mcp-proxy@latest plugin-telemetry "${args[@]}" \ - /dev/null 2>&1 & ) >/dev/null 2>&1 - disown 2>/dev/null fi +printf '%s\n' "${args[@]}" | python3 "$scriptDir/lib/telemetry_enqueue.py" "$cdir" 2>/dev/null + exit 0 diff --git a/plugins/alibabacloud-core/openclaw.plugin.json b/plugins/alibabacloud-core/openclaw.plugin.json index d8fa85b..d2c380d 100644 --- a/plugins/alibabacloud-core/openclaw.plugin.json +++ b/plugins/alibabacloud-core/openclaw.plugin.json @@ -1,7 +1,7 @@ { "id": "alibabacloud-core", "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.", "configSchema": { "type": "object", diff --git a/plugins/alibabacloud-ecs-ops/hooks/scripts/lib/post_handler.py b/plugins/alibabacloud-ecs-ops/hooks/scripts/lib/post_handler.py index f4c27eb..5416c5c 100644 --- a/plugins/alibabacloud-ecs-ops/hooks/scripts/lib/post_handler.py +++ b/plugins/alibabacloud-ecs-ops/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 + plugin-telemetry (queued via bounded worker) Exit codes: 0 — args printed (caller should upload) diff --git a/plugins/alibabacloud-ecs-ops/hooks/scripts/lib/stop_handler.py b/plugins/alibabacloud-ecs-ops/hooks/scripts/lib/stop_handler.py index 3e31534..36716bc 100644 --- a/plugins/alibabacloud-ecs-ops/hooks/scripts/lib/stop_handler.py +++ b/plugins/alibabacloud-ecs-ops/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 = { diff --git a/plugins/alibabacloud-ecs-ops/hooks/scripts/lib/telemetry_enqueue.py b/plugins/alibabacloud-ecs-ops/hooks/scripts/lib/telemetry_enqueue.py new file mode 100644 index 0000000..034ebd4 --- /dev/null +++ b/plugins/alibabacloud-ecs-ops/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 [--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() diff --git a/plugins/alibabacloud-ecs-ops/hooks/scripts/lib/telemetry_worker.py b/plugins/alibabacloud-ecs-ops/hooks/scripts/lib/telemetry_worker.py new file mode 100644 index 0000000..5e37e27 --- /dev/null +++ b/plugins/alibabacloud-ecs-ops/hooks/scripts/lib/telemetry_worker.py @@ -0,0 +1,429 @@ +#!/usr/bin/env python3 +"""Bounded telemetry upload worker. + +Processes queued telemetry events with: +- Single-instance control via file lock (flock) +- Fixed virtualenv with pinned package (no per-event uvx resolution) +- Configurable concurrency limit (sequential batch processing) +- Hard timeout per upload with subprocess reaping +- Finite retry budget with dead-letter logging +- No orphan processes (proper session group cleanup) + +Usage: + python3 telemetry_worker.py [--cleanup-only] +""" + +import fcntl +import json +import os +import signal +import subprocess +import sys +import time + +MCP_PROXY_PACKAGE = "alibabacloud.mcp-proxy" +MCP_PROXY_PINNED_VERSION = "0.5.1" +MCP_PROXY_PIN = f"{MCP_PROXY_PACKAGE}=={MCP_PROXY_PINNED_VERSION}" + +DEFAULT_MAX_CONCURRENT = int(os.environ.get( + "ALIBABACLOUD_TELEMETRY_MAX_CONCURRENT", "4" +)) +DEFAULT_HARD_TIMEOUT = int(os.environ.get( + "ALIBABACLOUD_TELEMETRY_UPLOAD_TIMEOUT", "30" +)) +DEFAULT_MAX_RETRIES = int(os.environ.get( + "ALIBABACLOUD_TELEMETRY_MAX_RETRIES", "2" +)) +MAX_QUEUE_SIZE = int(os.environ.get( + "ALIBABACLOUD_TELEMETRY_MAX_QUEUE", "500" +)) +FAILED_RETENTION_DAYS = 7 +PENDING_STALE_HOURS = 24 + +_debug_log_path = None + + +def _log(msg): + ts = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + line = f"[{ts}] {msg}" + print(line, flush=True) + if _debug_log_path: + try: + with open(_debug_log_path, "a") as f: + f.write(line + "\n") + except OSError: + pass + + +def _acquire_lock(state_dir): + """Try to acquire exclusive lock. Returns (lock_fd, lock_path) or None.""" + os.makedirs(state_dir, exist_ok=True) + lock_path = os.path.join(state_dir, "telemetry-worker.lock") + try: + fd = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o600) + except OSError: + return None + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + return fd, lock_path + except (IOError, OSError): + os.close(fd) + return None + + +def _release_lock(lock_info): + if lock_info is None: + return + fd, _ = lock_info + try: + fcntl.flock(fd, fcntl.LOCK_UN) + os.close(fd) + except OSError: + pass + + +def _is_pid_alive(pid): + try: + os.kill(pid, 0) + return True + except (OSError, ProcessLookupError): + return False + + +def _reap_pid(pid): + """Reap a zombie child if it is ours.""" + try: + os.waitpid(pid, os.WNOHANG) + except ChildProcessError: + pass + + +def _write_pid_file(state_dir): + pid_path = os.path.join(state_dir, "telemetry-worker.pid") + with open(pid_path, "w") as f: + f.write(str(os.getpid())) + return pid_path + + +def _check_existing_worker(state_dir): + """Return True if another worker is already running.""" + pid_path = os.path.join(state_dir, "telemetry-worker.pid") + try: + with open(pid_path) as f: + pid = int(f.read().strip()) + if _is_pid_alive(pid) and pid != os.getpid(): + return True + except (FileNotFoundError, ValueError, OSError): + pass + return False + + +def _remove_pid_file(pid_path): + try: + with open(pid_path) as f: + if int(f.read().strip()) == os.getpid(): + os.unlink(pid_path) + except (FileNotFoundError, ValueError, OSError): + pass + + +def _get_upload_cmd(state_dir): + """Return the upload command argv prefix using fixed venv or pinned uvx.""" + override = os.environ.get("ALIBABACLOUD_TELEMETRY_UPLOADER") + if override: + return override.split() + + venv_dir = os.path.join(state_dir, ".venv") + venv_bin = os.path.join(venv_dir, "bin", "plugin-telemetry") + + if os.path.isfile(venv_bin): + return [venv_bin] + + if _ensure_venv(state_dir): + if os.path.isfile(venv_bin): + return [venv_bin] + + return [ + "uvx", "--from", + f"{MCP_PROXY_PACKAGE}=={MCP_PROXY_PINNED_VERSION}", + "plugin-telemetry", + ] + + +def _ensure_venv(state_dir): + """Create or update the fixed virtualenv. Returns True on success.""" + venv_dir = os.path.join(state_dir, ".venv") + marker = os.path.join(venv_dir, ".telemetry-version") + + if os.path.isfile(marker): + try: + with open(marker) as f: + if f.read().strip() == MCP_PROXY_PIN: + return True + except OSError: + pass + + try: + import venv as venv_mod + _log(f"Creating venv at {venv_dir}") + venv_mod.EnvBuilder(with_pip=True, clear=True).create(venv_dir) + except Exception as e: + _log(f"venv creation failed: {e}") + return False + + pip_bin = os.path.join(venv_dir, "bin", "pip") + try: + result = subprocess.run( + [pip_bin, "install", "--quiet", MCP_PROXY_PIN], + timeout=120, + capture_output=True, + text=True, + ) + if result.returncode != 0: + _log(f"pip install failed: {result.stderr[:500]}") + return False + except subprocess.TimeoutExpired: + _log("pip install timed out after 120s") + return False + except Exception as e: + _log(f"pip install error: {e}") + return False + + try: + with open(marker, "w") as f: + f.write(MCP_PROXY_PIN) + except OSError: + pass + + _log(f"Installed {MCP_PROXY_PIN}") + return True + + +def _upload_one(cmd_prefix, args_dict): + """Run a single upload with hard timeout. Raises on failure.""" + argv = list(cmd_prefix) + for key, value in args_dict.items(): + if value is not None and value != "": + argv.append(f"--{key}") + argv.append(str(value)) + + timeout = DEFAULT_HARD_TIMEOUT + + proc = subprocess.Popen( + argv, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + start_new_session=True, + ) + + try: + stdout, stderr = proc.communicate(timeout=timeout) + if proc.returncode != 0: + err_snippet = (stderr.decode("utf-8", errors="replace"))[:300] + raise RuntimeError( + f"upload exited {proc.returncode}: {err_snippet}" + ) + except subprocess.TimeoutExpired: + _kill_process_tree(proc) + raise RuntimeError(f"upload timed out after {timeout}s") + + +def _kill_process_tree(proc): + """Kill a subprocess and its entire process group. No orphans.""" + try: + pgid = os.getpgid(proc.pid) + os.killpg(pgid, signal.SIGTERM) + except (ProcessLookupError, OSError): + pass + + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + try: + pgid = os.getpgid(proc.pid) + os.killpg(pgid, signal.SIGKILL) + except (ProcessLookupError, OSError): + pass + try: + proc.kill() + except (ProcessLookupError, OSError): + pass + try: + proc.wait(timeout=3) + except subprocess.TimeoutExpired: + pass + + _reap_pid(proc.pid) + + +def _list_pending(queue_dir): + """List pending event files sorted by timestamp, capped at MAX_QUEUE_SIZE.""" + try: + files = [ + f for f in os.listdir(queue_dir) + if f.endswith(".json") and not f.endswith(".tmp") + ] + except FileNotFoundError: + return [] + + files.sort() + if len(files) > MAX_QUEUE_SIZE: + excess = files[MAX_QUEUE_SIZE:] + for f in excess: + try: + os.unlink(os.path.join(queue_dir, f)) + except OSError: + pass + _log(f"Queue overflow: dropped {len(excess)} oldest events") + files = files[:MAX_QUEUE_SIZE] + return files + + +def _process_batch(state_dir): + """Process one batch of pending events. Returns count processed.""" + queue_dir = os.path.join(state_dir, "telemetry-queue", "pending") + failed_dir = os.path.join(state_dir, "telemetry-queue", "failed") + os.makedirs(failed_dir, mode=0o700, exist_ok=True) + + files = _list_pending(queue_dir) + if not files: + return 0 + + cmd_prefix = _get_upload_cmd(state_dir) + processed = 0 + + for filename in files: + filepath = os.path.join(queue_dir, filename) + try: + with open(filepath) as f: + event = json.load(f) + except (json.JSONDecodeError, OSError) as e: + _log(f"Removing corrupt event {filename}: {e}") + _safe_unlink(filepath) + continue + + args_dict = event.get("args", {}) + retries = event.get("retries", 0) + event_type = args_dict.get("event-type", "unknown") + + try: + _upload_one(cmd_prefix, args_dict) + _safe_unlink(filepath) + processed += 1 + except RuntimeError as e: + _log(f"Upload failed for {event_type} ({filename}): {e}") + if retries < DEFAULT_MAX_RETRIES: + event["retries"] = retries + 1 + _write_back(filepath, event) + else: + _move_to_failed(filepath, failed_dir, event) + _log(f"Moved to failed after {DEFAULT_MAX_RETRIES} retries: {filename}") + except Exception as e: + _log(f"Unexpected error processing {filename}: {e}") + _move_to_failed(filepath, failed_dir, event) + + return processed + + +def _safe_unlink(path): + try: + os.unlink(path) + except OSError: + pass + + +def _write_back(filepath, event): + try: + with open(filepath, "w") as f: + json.dump(event, f) + except OSError: + pass + + +def _move_to_failed(filepath, failed_dir, event): + try: + os.makedirs(failed_dir, mode=0o700, exist_ok=True) + dest = os.path.join(failed_dir, os.path.basename(filepath)) + with open(dest, "w") as f: + json.dump(event, f) + os.unlink(filepath) + except OSError: + _safe_unlink(filepath) + + +def _cleanup_old(state_dir): + """Remove old failed uploads and stale pending events.""" + now = time.time() + for subdir in ("failed", "pending"): + d = os.path.join(state_dir, "telemetry-queue", subdir) + if not os.path.isdir(d): + continue + max_age = ( + FAILED_RETENTION_DAYS * 86400 + if subdir == "failed" + else PENDING_STALE_HOURS * 3600 + ) + try: + for f in os.listdir(d): + fp = os.path.join(d, f) + try: + if now - os.path.getmtime(fp) > max_age: + os.unlink(fp) + except OSError: + pass + except FileNotFoundError: + pass + + +def main(): + if len(sys.argv) < 2: + print(f"Usage: {sys.argv[0]} [--cleanup-only]", file=sys.stderr) + sys.exit(1) + + state_dir = sys.argv[1] + cleanup_only = "--cleanup-only" in sys.argv + + global _debug_log_path + if os.environ.get("ALIBABACLOUD_TELEMETRY_WORKER_DEBUG") == "1": + _debug_log_path = os.path.join(state_dir, "worker-debug.log") + + if not os.path.isdir(state_dir): + _log(f"State dir does not exist: {state_dir}") + sys.exit(1) + + os.makedirs( + os.path.join(state_dir, "telemetry-queue", "pending"), + mode=0o700, exist_ok=True, + ) + + if _check_existing_worker(state_dir): + return + + lock_info = _acquire_lock(state_dir) + if lock_info is None: + return + + pid_path = _write_pid_file(state_dir) + try: + if cleanup_only: + _cleanup_old(state_dir) + return + + iterations = 0 + while True: + count = _process_batch(state_dir) + iterations += 1 + if count == 0: + break + if iterations >= 1000: + _log("Safety limit reached (1000 iterations), exiting") + break + _cleanup_old(state_dir) + finally: + _remove_pid_file(pid_path) + _release_lock(lock_info) + + +if __name__ == "__main__": + main() diff --git a/plugins/alibabacloud-ecs-ops/hooks/scripts/post-tool-trace.sh b/plugins/alibabacloud-ecs-ops/hooks/scripts/post-tool-trace.sh index 6f3ae4f..184b59c 100755 --- a/plugins/alibabacloud-ecs-ops/hooks/scripts/post-tool-trace.sh +++ b/plugins/alibabacloud-ecs-ops/hooks/scripts/post-tool-trace.sh @@ -1,10 +1,11 @@ #!/bin/bash # Post-tool-use hook wrapper. Delegates classification + status detection to -# lib/post_handler.py, then fires `uvx alibabacloud.mcp-proxy@latest -# plugin-telemetry` in the background. Always returns success to the agent. +# lib/post_handler.py, then queues the event for bounded background upload. +# Always returns success to the agent. set +e umask 077 + return_success() { echo '{"continue":true}' exit 0 @@ -83,6 +84,15 @@ payload=$(head -c 65536) client=$(detect_client_bash "$payload") cdir=$(state_dir_for_client "$client") +# Dump raw payload to debug.log for diagnosis +if [ "${ALIBABACLOUD_TELEMETRY_DEBUG}" = "1" ]; then + { + printf '[%s] [post-tool] raw-payload (%d bytes):\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" "${#payload}" + printf '%s\n' "$payload" | head -c 4096 + printf '\n---end-payload---\n' + } >> "$cdir/debug.log" 2>/dev/null +fi + # Optional raw-payload trace: dump full stdin to a file so future bugs # can be diagnosed without guessing at the payload shape. if [ "${ALIBABACLOUD_TELEMETRY_TRACE_PAYLOAD}" = "1" ]; then @@ -137,7 +147,7 @@ done # Dry-run mode: log instead of upload if [ "${ALIBABACLOUD_TELEMETRY_DRY_RUN}" = "1" ]; then { - printf 'DRYRUN: uvx alibabacloud.mcp-proxy@latest plugin-telemetry' + printf 'DRYRUN: queue telemetry event' for a in "${args[@]}"; do printf ' %q' "$a" done @@ -147,10 +157,21 @@ if [ "${ALIBABACLOUD_TELEMETRY_DRY_RUN}" = "1" ]; then return_success fi -# Fire-and-forget: detach so the agent loop never waits on uvx. +# Queue-based upload: write event args to a JSON queue file and start +# the bounded worker. Replaces the old fire-and-forget +# `uvx alibabacloud.mcp-proxy@latest` pattern that spawned one +# unbounded process per event, causing orphan process accumulation. debug_log "$cdir" "decision=upload event=$(extract_arg --event-type "${args[@]}") tool=$(extract_arg --tool-name "${args[@]}")" -( uvx alibabacloud.mcp-proxy@latest plugin-telemetry "${args[@]}" \ - /dev/null 2>&1 & ) >/dev/null 2>&1 -disown 2>/dev/null + +if [ "${ALIBABACLOUD_TELEMETRY_DEBUG}" = "1" ]; then + { + printf '[%s] [post-tool] enqueue event=%s tool=%s\n' \ + "$(date -u +%Y%m%dT%H%M%SZ)" \ + "$(extract_arg --event-type "${args[@]}")" \ + "$(extract_arg --tool-name "${args[@]}")" + } >> "$cdir/debug.log" 2>/dev/null +fi + +printf '%s\n' "${args[@]}" | python3 "$scriptDir/lib/telemetry_enqueue.py" "$cdir" 2>/dev/null return_success diff --git a/plugins/alibabacloud-ecs-ops/hooks/scripts/prompt-trace.sh b/plugins/alibabacloud-ecs-ops/hooks/scripts/prompt-trace.sh index d08185a..2e33a87 100755 --- a/plugins/alibabacloud-ecs-ops/hooks/scripts/prompt-trace.sh +++ b/plugins/alibabacloud-ecs-ops/hooks/scripts/prompt-trace.sh @@ -1,12 +1,13 @@ #!/bin/bash # UserPromptSubmit hook wrapper. Detects slash-style skill invocations # (`/alibabacloud-*: ...`) submitted directly to Claude Code as -# prompts. Delegates classification to lib/prompt_handler.py, then fires -# `uvx alibabacloud.mcp-proxy@latest plugin-telemetry` in the background. +# prompts. Delegates classification to lib/prompt_handler.py, then queues +# the event for bounded background upload. # Always returns success to the agent. set +e umask 077 + return_success() { echo '{"continue":true}' exit 0 @@ -85,6 +86,15 @@ payload=$(head -c 65536) client=$(detect_client_bash "$payload") cdir=$(state_dir_for_client "$client") +# Dump raw payload to debug.log for diagnosis +if [ "${ALIBABACLOUD_TELEMETRY_DEBUG}" = "1" ]; then + { + printf '[%s] [prompt] raw-payload (%d bytes):\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" "${#payload}" + printf '%s\n' "$payload" | head -c 4096 + printf '\n---end-payload---\n' + } >> "$cdir/debug.log" 2>/dev/null +fi + # Optional raw-payload trace: dump full stdin to a file so future bugs # can be diagnosed without guessing at the payload shape. if [ "${ALIBABACLOUD_TELEMETRY_TRACE_PAYLOAD}" = "1" ]; then @@ -140,7 +150,7 @@ done # Dry-run mode: log instead of upload if [ "${ALIBABACLOUD_TELEMETRY_DRY_RUN}" = "1" ]; then { - printf 'DRYRUN: uvx alibabacloud.mcp-proxy@latest plugin-telemetry' + printf 'DRYRUN: queue telemetry event' for a in "${args[@]}"; do printf ' %q' "$a" done @@ -150,10 +160,20 @@ if [ "${ALIBABACLOUD_TELEMETRY_DRY_RUN}" = "1" ]; then return_success fi -# Fire-and-forget: detach so the agent loop never waits on uvx. +# Queue-based upload: write event args to a JSON queue file and start +# the bounded worker. Replaces the old fire-and-forget +# `uvx alibabacloud.mcp-proxy@latest` pattern that spawned one +# unbounded process per event, causing orphan process accumulation. debug_log "$cdir" "decision=upload skill=$(extract_arg --skill-name "${args[@]}")" -( uvx alibabacloud.mcp-proxy@latest plugin-telemetry "${args[@]}" \ - /dev/null 2>&1 & ) >/dev/null 2>&1 -disown 2>/dev/null + +if [ "${ALIBABACLOUD_TELEMETRY_DEBUG}" = "1" ]; then + { + printf '[%s] [prompt] enqueue skill=%s\n' \ + "$(date -u +%Y%m%dT%H%M%SZ)" \ + "$(extract_arg --skill-name "${args[@]}")" + } >> "$cdir/debug.log" 2>/dev/null +fi + +printf '%s\n' "${args[@]}" | python3 "$scriptDir/lib/telemetry_enqueue.py" "$cdir" 2>/dev/null return_success diff --git a/plugins/alibabacloud-ecs-ops/hooks/scripts/stop-turn-increment.sh b/plugins/alibabacloud-ecs-ops/hooks/scripts/stop-turn-increment.sh index 9cc6d00..ed734b3 100755 --- a/plugins/alibabacloud-ecs-ops/hooks/scripts/stop-turn-increment.sh +++ b/plugins/alibabacloud-ecs-ops/hooks/scripts/stop-turn-increment.sh @@ -3,11 +3,12 @@ # Turn number is consumed by post-tool-trace.sh to tag --turn on each event. # Also bound to StopFailure for symmetry; both paths log identically. # When the turn involved alibabacloud tools, stop_handler.py emits a -# user_prompt_turn_start event to stdout which we upload to remote telemetry. +# user_prompt_turn_start event to stdout which we queue for bounded upload. # Delegates to lib/stop_handler.py which uses fcntl-locked per-session state. set +e umask 077 + if [ "${ALIBABACLOUD_TELEMETRY}" = "false" ]; then exit 0 fi @@ -74,6 +75,15 @@ payload=$(head -c 65536) client=$(detect_client_bash "$payload") cdir=$(state_dir_for_client "$client") +# Dump raw payload to debug.log for diagnosis +if [ "${ALIBABACLOUD_TELEMETRY_DEBUG}" = "1" ]; then + { + printf '[%s] [stop] raw-payload (%d bytes):\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" "${#payload}" + printf '%s\n' "$payload" | head -c 4096 + printf '\n---end-payload---\n' + } >> "$cdir/debug.log" 2>/dev/null +fi + if [ "${ALIBABACLOUD_TELEMETRY_TRACE_PAYLOAD}" = "1" ]; then payloadDir="$cdir/raw-payloads" mkdir -p "$payloadDir" 2>/dev/null && chmod 700 "$payloadDir" 2>/dev/null @@ -122,7 +132,7 @@ done # Dry-run mode: log instead of upload if [ "${ALIBABACLOUD_TELEMETRY_DRY_RUN}" = "1" ]; then { - printf 'DRYRUN: uvx alibabacloud.mcp-proxy@latest plugin-telemetry' + printf 'DRYRUN: queue telemetry event' for a in "${args[@]}"; do printf ' %q' "$a" done @@ -132,10 +142,20 @@ if [ "${ALIBABACLOUD_TELEMETRY_DRY_RUN}" = "1" ]; then exit 0 fi -# Fire-and-forget: detach so the agent loop never waits on uvx. +# Queue-based upload: write event args to a JSON queue file and start +# the bounded worker. Replaces the old fire-and-forget +# `uvx alibabacloud.mcp-proxy@latest` pattern that spawned one +# unbounded process per event, causing orphan process accumulation. debug_log "$cdir" "[stop] decision=upload event=$(extract_arg --event-type "${args[@]}")" -( uvx alibabacloud.mcp-proxy@latest plugin-telemetry "${args[@]}" \ - /dev/null 2>&1 & ) >/dev/null 2>&1 -disown 2>/dev/null + +if [ "${ALIBABACLOUD_TELEMETRY_DEBUG}" = "1" ]; then + { + printf '[%s] [stop] enqueue event=%s\n' \ + "$(date -u +%Y%m%dT%H%M%SZ)" \ + "$(extract_arg --event-type "${args[@]}")" + } >> "$cdir/debug.log" 2>/dev/null +fi + +printf '%s\n' "${args[@]}" | python3 "$scriptDir/lib/telemetry_enqueue.py" "$cdir" 2>/dev/null exit 0 diff --git a/plugins/alibabacloud-spec-ops/hooks/scripts/lib/post_handler.py b/plugins/alibabacloud-spec-ops/hooks/scripts/lib/post_handler.py index 1fe8caa..81a3680 100644 --- a/plugins/alibabacloud-spec-ops/hooks/scripts/lib/post_handler.py +++ b/plugins/alibabacloud-spec-ops/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 + plugin-telemetry (queued via bounded worker) Exit codes: 0 — args printed (caller should upload) diff --git a/plugins/alibabacloud-spec-ops/hooks/scripts/lib/stop_handler.py b/plugins/alibabacloud-spec-ops/hooks/scripts/lib/stop_handler.py index 3e31534..36716bc 100644 --- a/plugins/alibabacloud-spec-ops/hooks/scripts/lib/stop_handler.py +++ b/plugins/alibabacloud-spec-ops/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 = { diff --git a/plugins/alibabacloud-spec-ops/hooks/scripts/lib/telemetry_enqueue.py b/plugins/alibabacloud-spec-ops/hooks/scripts/lib/telemetry_enqueue.py new file mode 100644 index 0000000..034ebd4 --- /dev/null +++ b/plugins/alibabacloud-spec-ops/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 [--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() diff --git a/plugins/alibabacloud-spec-ops/hooks/scripts/lib/telemetry_worker.py b/plugins/alibabacloud-spec-ops/hooks/scripts/lib/telemetry_worker.py new file mode 100644 index 0000000..5e37e27 --- /dev/null +++ b/plugins/alibabacloud-spec-ops/hooks/scripts/lib/telemetry_worker.py @@ -0,0 +1,429 @@ +#!/usr/bin/env python3 +"""Bounded telemetry upload worker. + +Processes queued telemetry events with: +- Single-instance control via file lock (flock) +- Fixed virtualenv with pinned package (no per-event uvx resolution) +- Configurable concurrency limit (sequential batch processing) +- Hard timeout per upload with subprocess reaping +- Finite retry budget with dead-letter logging +- No orphan processes (proper session group cleanup) + +Usage: + python3 telemetry_worker.py [--cleanup-only] +""" + +import fcntl +import json +import os +import signal +import subprocess +import sys +import time + +MCP_PROXY_PACKAGE = "alibabacloud.mcp-proxy" +MCP_PROXY_PINNED_VERSION = "0.5.1" +MCP_PROXY_PIN = f"{MCP_PROXY_PACKAGE}=={MCP_PROXY_PINNED_VERSION}" + +DEFAULT_MAX_CONCURRENT = int(os.environ.get( + "ALIBABACLOUD_TELEMETRY_MAX_CONCURRENT", "4" +)) +DEFAULT_HARD_TIMEOUT = int(os.environ.get( + "ALIBABACLOUD_TELEMETRY_UPLOAD_TIMEOUT", "30" +)) +DEFAULT_MAX_RETRIES = int(os.environ.get( + "ALIBABACLOUD_TELEMETRY_MAX_RETRIES", "2" +)) +MAX_QUEUE_SIZE = int(os.environ.get( + "ALIBABACLOUD_TELEMETRY_MAX_QUEUE", "500" +)) +FAILED_RETENTION_DAYS = 7 +PENDING_STALE_HOURS = 24 + +_debug_log_path = None + + +def _log(msg): + ts = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + line = f"[{ts}] {msg}" + print(line, flush=True) + if _debug_log_path: + try: + with open(_debug_log_path, "a") as f: + f.write(line + "\n") + except OSError: + pass + + +def _acquire_lock(state_dir): + """Try to acquire exclusive lock. Returns (lock_fd, lock_path) or None.""" + os.makedirs(state_dir, exist_ok=True) + lock_path = os.path.join(state_dir, "telemetry-worker.lock") + try: + fd = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o600) + except OSError: + return None + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + return fd, lock_path + except (IOError, OSError): + os.close(fd) + return None + + +def _release_lock(lock_info): + if lock_info is None: + return + fd, _ = lock_info + try: + fcntl.flock(fd, fcntl.LOCK_UN) + os.close(fd) + except OSError: + pass + + +def _is_pid_alive(pid): + try: + os.kill(pid, 0) + return True + except (OSError, ProcessLookupError): + return False + + +def _reap_pid(pid): + """Reap a zombie child if it is ours.""" + try: + os.waitpid(pid, os.WNOHANG) + except ChildProcessError: + pass + + +def _write_pid_file(state_dir): + pid_path = os.path.join(state_dir, "telemetry-worker.pid") + with open(pid_path, "w") as f: + f.write(str(os.getpid())) + return pid_path + + +def _check_existing_worker(state_dir): + """Return True if another worker is already running.""" + pid_path = os.path.join(state_dir, "telemetry-worker.pid") + try: + with open(pid_path) as f: + pid = int(f.read().strip()) + if _is_pid_alive(pid) and pid != os.getpid(): + return True + except (FileNotFoundError, ValueError, OSError): + pass + return False + + +def _remove_pid_file(pid_path): + try: + with open(pid_path) as f: + if int(f.read().strip()) == os.getpid(): + os.unlink(pid_path) + except (FileNotFoundError, ValueError, OSError): + pass + + +def _get_upload_cmd(state_dir): + """Return the upload command argv prefix using fixed venv or pinned uvx.""" + override = os.environ.get("ALIBABACLOUD_TELEMETRY_UPLOADER") + if override: + return override.split() + + venv_dir = os.path.join(state_dir, ".venv") + venv_bin = os.path.join(venv_dir, "bin", "plugin-telemetry") + + if os.path.isfile(venv_bin): + return [venv_bin] + + if _ensure_venv(state_dir): + if os.path.isfile(venv_bin): + return [venv_bin] + + return [ + "uvx", "--from", + f"{MCP_PROXY_PACKAGE}=={MCP_PROXY_PINNED_VERSION}", + "plugin-telemetry", + ] + + +def _ensure_venv(state_dir): + """Create or update the fixed virtualenv. Returns True on success.""" + venv_dir = os.path.join(state_dir, ".venv") + marker = os.path.join(venv_dir, ".telemetry-version") + + if os.path.isfile(marker): + try: + with open(marker) as f: + if f.read().strip() == MCP_PROXY_PIN: + return True + except OSError: + pass + + try: + import venv as venv_mod + _log(f"Creating venv at {venv_dir}") + venv_mod.EnvBuilder(with_pip=True, clear=True).create(venv_dir) + except Exception as e: + _log(f"venv creation failed: {e}") + return False + + pip_bin = os.path.join(venv_dir, "bin", "pip") + try: + result = subprocess.run( + [pip_bin, "install", "--quiet", MCP_PROXY_PIN], + timeout=120, + capture_output=True, + text=True, + ) + if result.returncode != 0: + _log(f"pip install failed: {result.stderr[:500]}") + return False + except subprocess.TimeoutExpired: + _log("pip install timed out after 120s") + return False + except Exception as e: + _log(f"pip install error: {e}") + return False + + try: + with open(marker, "w") as f: + f.write(MCP_PROXY_PIN) + except OSError: + pass + + _log(f"Installed {MCP_PROXY_PIN}") + return True + + +def _upload_one(cmd_prefix, args_dict): + """Run a single upload with hard timeout. Raises on failure.""" + argv = list(cmd_prefix) + for key, value in args_dict.items(): + if value is not None and value != "": + argv.append(f"--{key}") + argv.append(str(value)) + + timeout = DEFAULT_HARD_TIMEOUT + + proc = subprocess.Popen( + argv, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + start_new_session=True, + ) + + try: + stdout, stderr = proc.communicate(timeout=timeout) + if proc.returncode != 0: + err_snippet = (stderr.decode("utf-8", errors="replace"))[:300] + raise RuntimeError( + f"upload exited {proc.returncode}: {err_snippet}" + ) + except subprocess.TimeoutExpired: + _kill_process_tree(proc) + raise RuntimeError(f"upload timed out after {timeout}s") + + +def _kill_process_tree(proc): + """Kill a subprocess and its entire process group. No orphans.""" + try: + pgid = os.getpgid(proc.pid) + os.killpg(pgid, signal.SIGTERM) + except (ProcessLookupError, OSError): + pass + + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + try: + pgid = os.getpgid(proc.pid) + os.killpg(pgid, signal.SIGKILL) + except (ProcessLookupError, OSError): + pass + try: + proc.kill() + except (ProcessLookupError, OSError): + pass + try: + proc.wait(timeout=3) + except subprocess.TimeoutExpired: + pass + + _reap_pid(proc.pid) + + +def _list_pending(queue_dir): + """List pending event files sorted by timestamp, capped at MAX_QUEUE_SIZE.""" + try: + files = [ + f for f in os.listdir(queue_dir) + if f.endswith(".json") and not f.endswith(".tmp") + ] + except FileNotFoundError: + return [] + + files.sort() + if len(files) > MAX_QUEUE_SIZE: + excess = files[MAX_QUEUE_SIZE:] + for f in excess: + try: + os.unlink(os.path.join(queue_dir, f)) + except OSError: + pass + _log(f"Queue overflow: dropped {len(excess)} oldest events") + files = files[:MAX_QUEUE_SIZE] + return files + + +def _process_batch(state_dir): + """Process one batch of pending events. Returns count processed.""" + queue_dir = os.path.join(state_dir, "telemetry-queue", "pending") + failed_dir = os.path.join(state_dir, "telemetry-queue", "failed") + os.makedirs(failed_dir, mode=0o700, exist_ok=True) + + files = _list_pending(queue_dir) + if not files: + return 0 + + cmd_prefix = _get_upload_cmd(state_dir) + processed = 0 + + for filename in files: + filepath = os.path.join(queue_dir, filename) + try: + with open(filepath) as f: + event = json.load(f) + except (json.JSONDecodeError, OSError) as e: + _log(f"Removing corrupt event {filename}: {e}") + _safe_unlink(filepath) + continue + + args_dict = event.get("args", {}) + retries = event.get("retries", 0) + event_type = args_dict.get("event-type", "unknown") + + try: + _upload_one(cmd_prefix, args_dict) + _safe_unlink(filepath) + processed += 1 + except RuntimeError as e: + _log(f"Upload failed for {event_type} ({filename}): {e}") + if retries < DEFAULT_MAX_RETRIES: + event["retries"] = retries + 1 + _write_back(filepath, event) + else: + _move_to_failed(filepath, failed_dir, event) + _log(f"Moved to failed after {DEFAULT_MAX_RETRIES} retries: {filename}") + except Exception as e: + _log(f"Unexpected error processing {filename}: {e}") + _move_to_failed(filepath, failed_dir, event) + + return processed + + +def _safe_unlink(path): + try: + os.unlink(path) + except OSError: + pass + + +def _write_back(filepath, event): + try: + with open(filepath, "w") as f: + json.dump(event, f) + except OSError: + pass + + +def _move_to_failed(filepath, failed_dir, event): + try: + os.makedirs(failed_dir, mode=0o700, exist_ok=True) + dest = os.path.join(failed_dir, os.path.basename(filepath)) + with open(dest, "w") as f: + json.dump(event, f) + os.unlink(filepath) + except OSError: + _safe_unlink(filepath) + + +def _cleanup_old(state_dir): + """Remove old failed uploads and stale pending events.""" + now = time.time() + for subdir in ("failed", "pending"): + d = os.path.join(state_dir, "telemetry-queue", subdir) + if not os.path.isdir(d): + continue + max_age = ( + FAILED_RETENTION_DAYS * 86400 + if subdir == "failed" + else PENDING_STALE_HOURS * 3600 + ) + try: + for f in os.listdir(d): + fp = os.path.join(d, f) + try: + if now - os.path.getmtime(fp) > max_age: + os.unlink(fp) + except OSError: + pass + except FileNotFoundError: + pass + + +def main(): + if len(sys.argv) < 2: + print(f"Usage: {sys.argv[0]} [--cleanup-only]", file=sys.stderr) + sys.exit(1) + + state_dir = sys.argv[1] + cleanup_only = "--cleanup-only" in sys.argv + + global _debug_log_path + if os.environ.get("ALIBABACLOUD_TELEMETRY_WORKER_DEBUG") == "1": + _debug_log_path = os.path.join(state_dir, "worker-debug.log") + + if not os.path.isdir(state_dir): + _log(f"State dir does not exist: {state_dir}") + sys.exit(1) + + os.makedirs( + os.path.join(state_dir, "telemetry-queue", "pending"), + mode=0o700, exist_ok=True, + ) + + if _check_existing_worker(state_dir): + return + + lock_info = _acquire_lock(state_dir) + if lock_info is None: + return + + pid_path = _write_pid_file(state_dir) + try: + if cleanup_only: + _cleanup_old(state_dir) + return + + iterations = 0 + while True: + count = _process_batch(state_dir) + iterations += 1 + if count == 0: + break + if iterations >= 1000: + _log("Safety limit reached (1000 iterations), exiting") + break + _cleanup_old(state_dir) + finally: + _remove_pid_file(pid_path) + _release_lock(lock_info) + + +if __name__ == "__main__": + main() diff --git a/plugins/alibabacloud-spec-ops/hooks/scripts/post-tool-trace.sh b/plugins/alibabacloud-spec-ops/hooks/scripts/post-tool-trace.sh index f918e87..184b59c 100755 --- a/plugins/alibabacloud-spec-ops/hooks/scripts/post-tool-trace.sh +++ b/plugins/alibabacloud-spec-ops/hooks/scripts/post-tool-trace.sh @@ -1,7 +1,7 @@ #!/bin/bash # Post-tool-use hook wrapper. Delegates classification + status detection to -# lib/post_handler.py, then fires `uvx alibabacloud.mcp-proxy@latest -# plugin-telemetry` in the background. Always returns success to the agent. +# lib/post_handler.py, then queues the event for bounded background upload. +# Always returns success to the agent. set +e umask 077 @@ -147,7 +147,7 @@ done # Dry-run mode: log instead of upload if [ "${ALIBABACLOUD_TELEMETRY_DRY_RUN}" = "1" ]; then { - printf 'DRYRUN: uvx alibabacloud.mcp-proxy@latest plugin-telemetry' + printf 'DRYRUN: queue telemetry event' for a in "${args[@]}"; do printf ' %q' "$a" done @@ -157,31 +157,21 @@ if [ "${ALIBABACLOUD_TELEMETRY_DRY_RUN}" = "1" ]; then return_success fi -# Fire-and-forget: detach so the agent loop never waits on uvx. +# Queue-based upload: write event args to a JSON queue file and start +# the bounded worker. Replaces the old fire-and-forget +# `uvx alibabacloud.mcp-proxy@latest` pattern that spawned one +# unbounded process per event, causing orphan process accumulation. debug_log "$cdir" "decision=upload event=$(extract_arg --event-type "${args[@]}") tool=$(extract_arg --tool-name "${args[@]}")" if [ "${ALIBABACLOUD_TELEMETRY_DEBUG}" = "1" ]; then - # Debug mode: capture uvx output for diagnosis instead of discarding { - printf '[%s] [post-tool] upload-start cmd=uvx alibabacloud.mcp-proxy@latest plugin-telemetry' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" - for a in "${args[@]}"; do printf ' %q' "$a"; done - printf '\n' + printf '[%s] [post-tool] enqueue event=%s tool=%s\n' \ + "$(date -u +%Y%m%dT%H%M%SZ)" \ + "$(extract_arg --event-type "${args[@]}")" \ + "$(extract_arg --tool-name "${args[@]}")" } >> "$cdir/debug.log" 2>/dev/null - ( - uvx_out=$(uvx alibabacloud.mcp-proxy@latest plugin-telemetry "${args[@]}" &1) - uvx_rc=$? - { - printf '[%s] [post-tool] upload-done rc=%d\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" "$uvx_rc" - if [ -n "$uvx_out" ]; then - printf '[%s] [post-tool] upload-output: %s\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" "$uvx_out" - fi - } >> "$cdir/debug.log" 2>/dev/null - ) & - disown 2>/dev/null -else - ( uvx alibabacloud.mcp-proxy@latest plugin-telemetry "${args[@]}" \ - /dev/null 2>&1 & ) >/dev/null 2>&1 - disown 2>/dev/null fi +printf '%s\n' "${args[@]}" | python3 "$scriptDir/lib/telemetry_enqueue.py" "$cdir" 2>/dev/null + return_success diff --git a/plugins/alibabacloud-spec-ops/hooks/scripts/prompt-trace.sh b/plugins/alibabacloud-spec-ops/hooks/scripts/prompt-trace.sh index 928151d..2e33a87 100755 --- a/plugins/alibabacloud-spec-ops/hooks/scripts/prompt-trace.sh +++ b/plugins/alibabacloud-spec-ops/hooks/scripts/prompt-trace.sh @@ -1,8 +1,8 @@ #!/bin/bash # UserPromptSubmit hook wrapper. Detects slash-style skill invocations # (`/alibabacloud-*: ...`) submitted directly to Claude Code as -# prompts. Delegates classification to lib/prompt_handler.py, then fires -# `uvx alibabacloud.mcp-proxy@latest plugin-telemetry` in the background. +# prompts. Delegates classification to lib/prompt_handler.py, then queues +# the event for bounded background upload. # Always returns success to the agent. set +e umask 077 @@ -150,7 +150,7 @@ done # Dry-run mode: log instead of upload if [ "${ALIBABACLOUD_TELEMETRY_DRY_RUN}" = "1" ]; then { - printf 'DRYRUN: uvx alibabacloud.mcp-proxy@latest plugin-telemetry' + printf 'DRYRUN: queue telemetry event' for a in "${args[@]}"; do printf ' %q' "$a" done @@ -160,31 +160,20 @@ if [ "${ALIBABACLOUD_TELEMETRY_DRY_RUN}" = "1" ]; then return_success fi -# Fire-and-forget: detach so the agent loop never waits on uvx. +# Queue-based upload: write event args to a JSON queue file and start +# the bounded worker. Replaces the old fire-and-forget +# `uvx alibabacloud.mcp-proxy@latest` pattern that spawned one +# unbounded process per event, causing orphan process accumulation. debug_log "$cdir" "decision=upload skill=$(extract_arg --skill-name "${args[@]}")" if [ "${ALIBABACLOUD_TELEMETRY_DEBUG}" = "1" ]; then - # Debug mode: capture uvx output for diagnosis instead of discarding { - printf '[%s] [prompt] upload-start cmd=uvx alibabacloud.mcp-proxy@latest plugin-telemetry' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" - for a in "${args[@]}"; do printf ' %q' "$a"; done - printf '\n' + printf '[%s] [prompt] enqueue skill=%s\n' \ + "$(date -u +%Y%m%dT%H%M%SZ)" \ + "$(extract_arg --skill-name "${args[@]}")" } >> "$cdir/debug.log" 2>/dev/null - ( - uvx_out=$(uvx alibabacloud.mcp-proxy@latest plugin-telemetry "${args[@]}" &1) - uvx_rc=$? - { - printf '[%s] [prompt] upload-done rc=%d\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" "$uvx_rc" - if [ -n "$uvx_out" ]; then - printf '[%s] [prompt] upload-output: %s\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" "$uvx_out" - fi - } >> "$cdir/debug.log" 2>/dev/null - ) & - disown 2>/dev/null -else - ( uvx alibabacloud.mcp-proxy@latest plugin-telemetry "${args[@]}" \ - /dev/null 2>&1 & ) >/dev/null 2>&1 - disown 2>/dev/null fi +printf '%s\n' "${args[@]}" | python3 "$scriptDir/lib/telemetry_enqueue.py" "$cdir" 2>/dev/null + return_success diff --git a/plugins/alibabacloud-spec-ops/hooks/scripts/stop-turn-increment.sh b/plugins/alibabacloud-spec-ops/hooks/scripts/stop-turn-increment.sh index 932d484..ed734b3 100755 --- a/plugins/alibabacloud-spec-ops/hooks/scripts/stop-turn-increment.sh +++ b/plugins/alibabacloud-spec-ops/hooks/scripts/stop-turn-increment.sh @@ -3,7 +3,7 @@ # Turn number is consumed by post-tool-trace.sh to tag --turn on each event. # Also bound to StopFailure for symmetry; both paths log identically. # When the turn involved alibabacloud tools, stop_handler.py emits a -# user_prompt_turn_start event to stdout which we upload to remote telemetry. +# user_prompt_turn_start event to stdout which we queue for bounded upload. # Delegates to lib/stop_handler.py which uses fcntl-locked per-session state. set +e umask 077 @@ -132,7 +132,7 @@ done # Dry-run mode: log instead of upload if [ "${ALIBABACLOUD_TELEMETRY_DRY_RUN}" = "1" ]; then { - printf 'DRYRUN: uvx alibabacloud.mcp-proxy@latest plugin-telemetry' + printf 'DRYRUN: queue telemetry event' for a in "${args[@]}"; do printf ' %q' "$a" done @@ -142,31 +142,20 @@ if [ "${ALIBABACLOUD_TELEMETRY_DRY_RUN}" = "1" ]; then exit 0 fi -# Fire-and-forget: detach so the agent loop never waits on uvx. +# Queue-based upload: write event args to a JSON queue file and start +# the bounded worker. Replaces the old fire-and-forget +# `uvx alibabacloud.mcp-proxy@latest` pattern that spawned one +# unbounded process per event, causing orphan process accumulation. debug_log "$cdir" "[stop] decision=upload event=$(extract_arg --event-type "${args[@]}")" if [ "${ALIBABACLOUD_TELEMETRY_DEBUG}" = "1" ]; then - # Debug mode: capture uvx output for diagnosis instead of discarding { - printf '[%s] [stop] upload-start cmd=uvx alibabacloud.mcp-proxy@latest plugin-telemetry' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" - for a in "${args[@]}"; do printf ' %q' "$a"; done - printf '\n' + printf '[%s] [stop] enqueue event=%s\n' \ + "$(date -u +%Y%m%dT%H%M%SZ)" \ + "$(extract_arg --event-type "${args[@]}")" } >> "$cdir/debug.log" 2>/dev/null - ( - uvx_out=$(uvx alibabacloud.mcp-proxy@latest plugin-telemetry "${args[@]}" &1) - uvx_rc=$? - { - printf '[%s] [stop] upload-done rc=%d\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" "$uvx_rc" - if [ -n "$uvx_out" ]; then - printf '[%s] [stop] upload-output: %s\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" "$uvx_out" - fi - } >> "$cdir/debug.log" 2>/dev/null - ) & - disown 2>/dev/null -else - ( uvx alibabacloud.mcp-proxy@latest plugin-telemetry "${args[@]}" \ - /dev/null 2>&1 & ) >/dev/null 2>&1 - disown 2>/dev/null fi +printf '%s\n' "${args[@]}" | python3 "$scriptDir/lib/telemetry_enqueue.py" "$cdir" 2>/dev/null + exit 0 diff --git a/tests/test_telemetry_bounded_upload.py b/tests/test_telemetry_bounded_upload.py new file mode 100644 index 0000000..a6262da --- /dev/null +++ b/tests/test_telemetry_bounded_upload.py @@ -0,0 +1,511 @@ +#!/usr/bin/env python3 +"""Tests for the bounded telemetry upload system. + +Validates: +- Off switch (ALIBABACLOUD_TELEMETRY=false): no upload processes spawned +- Queue-based enqueue: events written to queue directory correctly +- Single-instance lock: only one worker processes at a time +- Bounded concurrency: configurable limits on parallel uploads +- Hard timeout: uploads killed after timeout, subprocess reaped +- Finite retry: failed events retried then moved to dead-letter +- Process cleanup: no orphan processes after worker exits +- Worker uses fixed venv, not uvx @latest per event +""" +import fcntl +import json +import os +import signal +import subprocess +import sys +import tempfile +import time +import unittest +from unittest import mock + +sys.path.insert( + 0, + os.path.join( + os.path.dirname(os.path.abspath(__file__)), + "..", + "plugins", + "alibabacloud-core", + "hooks", + "scripts", + "lib", + ), +) + +import telemetry_enqueue +import telemetry_worker + + +class TestOffSwitch(unittest.TestCase): + """ALIBABACLOUD_TELEMETRY=false must prevent all upload activity.""" + + def test_off_switch_skips_enqueue(self): + """Shell scripts check ALIBABACLOUD_TELEMETRY=false before enqueue.""" + shell = os.path.join( + os.path.dirname(os.path.abspath(__file__)), + "..", + "plugins", + "alibabacloud-core", + "hooks", + "scripts", + "post-tool-trace.sh", + ) + with open(shell) as f: + content = f.read() + self.assertIn("ALIBABACLOUD_TELEMETRY}", content) + self.assertIn('"false"', content) + self.assertIn("return_success", content) + idx_off = content.index('"false"') + idx_enqueue = content.index("telemetry_enqueue") + self.assertLess(idx_off, idx_enqueue) + + def test_worker_respects_missing_queue(self): + """Worker exits cleanly when queue directory is empty.""" + with tempfile.TemporaryDirectory() as tmp: + queue_dir = os.path.join(tmp, "telemetry-queue", "pending") + os.makedirs(queue_dir) + result = subprocess.run( + [sys.executable, telemetry_worker.__file__, tmp], + capture_output=True, + timeout=10, + ) + self.assertEqual(result.returncode, 0) + + +class TestEnqueue(unittest.TestCase): + """Events are correctly written to the queue directory.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp() + + def tearDown(self): + import shutil + shutil.rmtree(self.tmp, ignore_errors=True) + + def test_enqueue_creates_queue_file(self): + with mock.patch.object( + telemetry_worker, "__file__", "/dev/null" + ): + with mock.patch( + "telemetry_enqueue._ensure_worker" + ): + telemetry_enqueue.enqueue_event( + self.tmp, + {"event-type": "mcp_tool_use", "tool-name": "Foo"}, + start_worker=False, + ) + + pending = os.path.join(self.tmp, "telemetry-queue", "pending") + files = [f for f in os.listdir(pending) if f.endswith(".json")] + self.assertEqual(len(files), 1) + + with open(os.path.join(pending, files[0])) as f: + event = json.load(f) + + self.assertEqual(event["args"]["event-type"], "mcp_tool_use") + self.assertEqual(event["args"]["tool-name"], "Foo") + self.assertEqual(event["retries"], 0) + self.assertIn("timestamp", event) + + def test_enqueue_multiple_events(self): + with mock.patch("telemetry_enqueue._ensure_worker"): + for i in range(10): + telemetry_enqueue.enqueue_event( + self.tmp, + {"event-type": "tool_call", "index": str(i)}, + start_worker=False, + ) + + pending = os.path.join(self.tmp, "telemetry-queue", "pending") + files = [f for f in os.listdir(pending) if f.endswith(".json")] + self.assertEqual(len(files), 10) + + def test_parse_lines_to_dict(self): + text = "--event-type\nmcp_tool_use\n--tool-name\nFoo\n--status\nsuccess\n" + result = telemetry_enqueue._parse_lines_to_dict(text) + self.assertEqual(result, { + "event-type": "mcp_tool_use", + "tool-name": "Foo", + "status": "success", + }) + + def test_parse_lines_empty(self): + result = telemetry_enqueue._parse_lines_to_dict("") + self.assertEqual(result, {}) + + def test_parse_lines_skips_non_flag(self): + text = "noise\n--key\nvalue\n" + result = telemetry_enqueue._parse_lines_to_dict(text) + self.assertEqual(result, {"key": "value"}) + + +class TestSingleInstanceLock(unittest.TestCase): + """Only one worker can hold the lock at a time.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp() + + def tearDown(self): + import shutil + shutil.rmtree(self.tmp, ignore_errors=True) + + def test_acquire_and_release_lock(self): + lock_info = telemetry_worker._acquire_lock(self.tmp) + self.assertIsNotNone(lock_info) + fd, lock_path = lock_info + self.assertTrue(os.path.exists(lock_path)) + telemetry_worker._release_lock(lock_info) + + def test_second_lock_fails(self): + lock1 = telemetry_worker._acquire_lock(self.tmp) + self.assertIsNotNone(lock1) + + lock2 = telemetry_worker._acquire_lock(self.tmp) + self.assertIsNone(lock2) + + telemetry_worker._release_lock(lock1) + + def test_lock_released_allows_reacquire(self): + lock1 = telemetry_worker._acquire_lock(self.tmp) + telemetry_worker._release_lock(lock1) + + lock2 = telemetry_worker._acquire_lock(self.tmp) + self.assertIsNotNone(lock2) + telemetry_worker._release_lock(lock2) + + +class TestWorkerBatchProcessing(unittest.TestCase): + """Worker processes queued events correctly.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp() + self.queue_dir = os.path.join( + self.tmp, "telemetry-queue", "pending" + ) + self.failed_dir = os.path.join( + self.tmp, "telemetry-queue", "failed" + ) + os.makedirs(self.queue_dir, exist_ok=True) + + def tearDown(self): + import shutil + shutil.rmtree(self.tmp, ignore_errors=True) + + def _write_event(self, args, retries=0): + filename = f"{int(time.time() * 1000000)}-{len(os.listdir(self.queue_dir))}.json" + filepath = os.path.join(self.queue_dir, filename) + event = {"args": args, "retries": retries, "timestamp": time.time()} + with open(filepath, "w") as f: + json.dump(event, f) + return filepath + + @mock.patch("telemetry_worker._upload_one") + def test_process_batch_success(self, mock_upload): + self._write_event({"event-type": "tool_call", "tool-name": "A"}) + self._write_event({"event-type": "tool_call", "tool-name": "B"}) + + with mock.patch.dict(os.environ, {"ALIBABACLOUD_TELEMETRY_UPLOADER": "echo"}): + count = telemetry_worker._process_batch(self.tmp) + + self.assertEqual(count, 2) + self.assertEqual(mock_upload.call_count, 2) + remaining = os.listdir(self.queue_dir) + self.assertEqual(len(remaining), 0) + + @mock.patch("telemetry_worker._upload_one") + def test_process_batch_retry_on_failure(self, mock_upload): + mock_upload.side_effect = RuntimeError("upload failed") + self._write_event({"event-type": "tool_call"}) + + count = telemetry_worker._process_batch(self.tmp) + + self.assertEqual(count, 0) + remaining = os.listdir(self.queue_dir) + self.assertEqual(len(remaining), 1) + + with open(os.path.join(self.queue_dir, remaining[0])) as f: + event = json.load(f) + self.assertEqual(event["retries"], 1) + + @mock.patch("telemetry_worker._upload_one") + def test_process_batch_move_to_failed_after_max_retries(self, mock_upload): + mock_upload.side_effect = RuntimeError("permanent failure") + self._write_event( + {"event-type": "tool_call"}, + retries=telemetry_worker.DEFAULT_MAX_RETRIES, + ) + + count = telemetry_worker._process_batch(self.tmp) + + self.assertEqual(count, 0) + remaining = os.listdir(self.queue_dir) + self.assertEqual(len(remaining), 0) + + failed = os.listdir(self.failed_dir) + self.assertEqual(len(failed), 1) + + @mock.patch("telemetry_worker._upload_one") + def test_process_batch_corrupt_event_removed(self, mock_upload): + filepath = os.path.join(self.queue_dir, "corrupt.json") + with open(filepath, "w") as f: + f.write("not valid json{{{") + + count = telemetry_worker._process_batch(self.tmp) + + self.assertEqual(count, 0) + self.assertEqual(mock_upload.call_count, 0) + remaining = os.listdir(self.queue_dir) + self.assertEqual(len(remaining), 0) + + def test_queue_overflow_drops_oldest(self): + old_max = telemetry_worker.MAX_QUEUE_SIZE + telemetry_worker.MAX_QUEUE_SIZE = 3 + try: + for i in range(5): + self._write_event({"event-type": "tool_call", "idx": str(i)}) + + files = telemetry_worker._list_pending(self.queue_dir) + self.assertEqual(len(files), 3) + finally: + telemetry_worker.MAX_QUEUE_SIZE = old_max + + +class TestUploadCommand(unittest.TestCase): + """Worker uses fixed version, not @latest.""" + + def test_get_upload_cmd_override(self): + with mock.patch.dict( + os.environ, {"ALIBABACLOUD_TELEMETRY_UPLOADER": "echo test"} + ): + cmd = telemetry_worker._get_upload_cmd("/tmp/test") + self.assertEqual(cmd, ["echo", "test"]) + + def test_get_upload_cmd_no_venv_uses_pinned_uvx(self): + with tempfile.TemporaryDirectory() as tmp: + with mock.patch.dict( + os.environ, {}, clear=True + ): + cmd = telemetry_worker._get_upload_cmd(tmp) + self.assertIn("--from", cmd) + pin_found = any( + f"=={telemetry_worker.MCP_PROXY_PINNED_VERSION}" in c + for c in cmd + ) + self.assertTrue(pin_found, f"No pinned version in cmd: {cmd}") + self.assertNotIn("@latest", " ".join(cmd)) + + def test_get_upload_cmd_uses_existing_venv(self): + with tempfile.TemporaryDirectory() as tmp: + venv_bin = os.path.join(tmp, ".venv", "bin", "plugin-telemetry") + os.makedirs(os.path.dirname(venv_bin), exist_ok=True) + with open(venv_bin, "w") as f: + f.write("#!/bin/bash\necho ok") + os.chmod(venv_bin, 0o755) + + cmd = telemetry_worker._get_upload_cmd(tmp) + + self.assertEqual(cmd, [venv_bin]) + + +class TestProcessTreeKilling(unittest.TestCase): + """Subprocess cleanup kills entire process group.""" + + def test_kill_process_tree_terminates(self): + proc = subprocess.Popen( + ["sleep", "60"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + self.assertTrue(proc.poll() is None) + + telemetry_worker._kill_process_tree(proc) + + self.assertIsNotNone(proc.returncode) + + def test_kill_process_tree_already_dead(self): + proc = subprocess.Popen( + ["true"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + proc.wait() + + telemetry_worker._kill_process_tree(proc) + + +class TestUploadTimeout(unittest.TestCase): + """Uploads that exceed timeout are killed.""" + + @mock.patch("telemetry_worker.DEFAULT_HARD_TIMEOUT", 1) + def test_timeout_kills_slow_upload(self): + cmd_prefix = ["sleep", "60"] + with self.assertRaises(RuntimeError) as ctx: + telemetry_worker._upload_one(cmd_prefix, {}) + self.assertIn("timed out", str(ctx.exception)) + + +class TestPIDFile(unittest.TestCase): + """PID file tracks worker process.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp() + + def tearDown(self): + import shutil + shutil.rmtree(self.tmp, ignore_errors=True) + + def test_write_and_check_pid(self): + pid_path = telemetry_worker._write_pid_file(self.tmp) + self.assertTrue(os.path.exists(pid_path)) + + with open(pid_path) as f: + pid = int(f.read().strip()) + self.assertEqual(pid, os.getpid()) + + def test_check_existing_worker_self(self): + telemetry_worker._write_pid_file(self.tmp) + self.assertFalse(telemetry_worker._check_existing_worker(self.tmp)) + + def test_check_existing_worker_dead_pid(self): + pid_path = os.path.join(self.tmp, "telemetry-worker.pid") + with open(pid_path, "w") as f: + f.write("99999999") + self.assertFalse(telemetry_worker._check_existing_worker(self.tmp)) + + def test_remove_pid_file(self): + pid_path = telemetry_worker._write_pid_file(self.tmp) + telemetry_worker._remove_pid_file(pid_path) + self.assertFalse(os.path.exists(pid_path)) + + +class TestCleanup(unittest.TestCase): + """Old files are cleaned up by the worker.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp() + + def tearDown(self): + import shutil + shutil.rmtree(self.tmp, ignore_errors=True) + + def test_cleanup_old_failed(self): + failed_dir = os.path.join( + self.tmp, "telemetry-queue", "failed" + ) + os.makedirs(failed_dir) + old_file = os.path.join(failed_dir, "old.json") + with open(old_file, "w") as f: + f.write("{}") + old_time = time.time() - (telemetry_worker.FAILED_RETENTION_DAYS + 1) * 86400 + os.utime(old_file, (old_time, old_time)) + + telemetry_worker._cleanup_old(self.tmp) + + self.assertFalse(os.path.exists(old_file)) + + def test_cleanup_keeps_recent(self): + failed_dir = os.path.join( + self.tmp, "telemetry-queue", "failed" + ) + os.makedirs(failed_dir) + recent_file = os.path.join(failed_dir, "recent.json") + with open(recent_file, "w") as f: + f.write("{}") + + telemetry_worker._cleanup_old(self.tmp) + + self.assertTrue(os.path.exists(recent_file)) + + +class TestNoUvxAtLatest(unittest.TestCase): + """Verify no script invokes uvx @latest at runtime.""" + + def _read_scripts(self): + base = os.path.join( + os.path.dirname(os.path.abspath(__file__)), + "..", + "plugins", + "alibabacloud-core", + "hooks", + "scripts", + ) + contents = {} + for name in os.listdir(base): + if name.endswith(".sh"): + with open(os.path.join(base, name)) as f: + contents[name] = f.read() + return contents + + def test_no_uvx_at_latest_invocation(self): + for name, content in self._read_scripts().items(): + for line in content.splitlines(): + stripped = line.strip() + if stripped.startswith("#"): + continue + if "DRYRUN" in stripped: + continue + self.assertNotIn( + "uvx alibabacloud.mcp-proxy@latest", + stripped, + f"{name} still invokes uvx @latest: {stripped}", + ) + + def test_no_disown_in_scripts(self): + for name, content in self._read_scripts().items(): + for line in content.splitlines(): + stripped = line.strip() + if stripped.startswith("#"): + continue + self.assertNotIn( + "disown", + stripped, + f"{name} still uses disown: {stripped}", + ) + + +class TestWorkerIntegration(unittest.TestCase): + """Integration test: enqueue events then run worker.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp() + self.queue_dir = os.path.join( + self.tmp, "telemetry-queue", "pending" + ) + os.makedirs(self.queue_dir, exist_ok=True) + + def tearDown(self): + import shutil + shutil.rmtree(self.tmp, ignore_errors=True) + + @mock.patch("telemetry_worker._upload_one") + def test_full_enqueue_and_process(self, mock_upload): + with mock.patch("telemetry_enqueue._ensure_worker"): + telemetry_enqueue.enqueue_event( + self.tmp, + {"event-type": "mcp_tool_use", "tool-name": "TestTool"}, + start_worker=False, + ) + telemetry_enqueue.enqueue_event( + self.tmp, + {"event-type": "skill_invocation", "skill-name": "TestSkill"}, + start_worker=False, + ) + + pending_files = os.listdir(self.queue_dir) + self.assertEqual(len(pending_files), 2) + + with mock.patch.dict(os.environ, {"ALIBABACLOUD_TELEMETRY_UPLOADER": "echo"}): + count = telemetry_worker._process_batch(self.tmp) + + self.assertEqual(count, 2) + self.assertEqual(mock_upload.call_count, 2) + remaining = os.listdir(self.queue_dir) + self.assertEqual(len(remaining), 0) + + +if __name__ == "__main__": + unittest.main()