${total_estimated_cost:.4f}
+ TOTAL ESTIMATED COST
+
+ Last Entry:
+ {escape_html(last_entry_timestamp_str)}
+
+ {escape_html(title)}: None
" + + list_items_html = "".join(f"No token summary available.
" + + return f""" +Prompt Tokens: {token_summary.get("prompt_tokens", "N/A")}
+Completion Tokens: {token_summary.get("completion_tokens", "N/A")}
+Total Tokens: {token_summary.get("total_tokens", "N/A")}
+Estimated Cost: ${token_summary.get("estimated_cost", 0.0):.6f}
+No models used summary available.
" + + rows_html = "" + for model_info in models_summary: + model_info = model_info or {} # Ensure model_info is not None + rows_html += f""" +| Name | +Calls | +Cost | +Prompt Tokens | +Completion Tokens | +
|---|
No LLM call details available.
" + + rows_html = "" + for call in llm_calls: + call = call or {} # Ensure call is not None + rows_html += f""" +| Model | +ID | +Finish Reason | +Prompt Tokens | +Completion Tokens | +Cost | +Timestamp | +
|---|
+ Query: + {query_text} +
+No latest interaction data to display.
' + history_entries_html = 'No interaction history to display.
' + else: + # Data is assumed to be oldest to newest from read_session_data + data_for_processing = list(all_session_data) # Make a copy + + latest_interaction_data = data_for_processing.pop() # Removes and returns the last item (newest) + project_name_header = escape_html(latest_interaction_data.get("project_name", "AIDER ANALYTICS")) # Get project name from latest interaction + + # Index 0 for latest, but color is overridden by use_special_color_bar + latest_interaction_display_html = generate_interaction_html(latest_interaction_data, 0, use_special_color_bar=True, special_color_class="blue") + + history_entries_html_parts = [] + if not data_for_processing: + history_entries_html = 'No further interaction history to display.
' + else: + # Iterate from newest to oldest for display for the rest of the history + for i, interaction_data in enumerate(reversed(data_for_processing)): + # i will be 0 for the newest in remaining, 1 for next, etc. + history_entries_html_parts.append(generate_interaction_html(interaction_data, i)) + history_entries_html = "\n".join(history_entries_html_parts) + if not history_entries_html_parts: # Should not happen if data_for_processing was not empty + history_entries_html = 'No further interaction history to display.
' + + + if not DASHBOARD_TEMPLATE_FILE.exists(): + print(f"Error: Dashboard template file not found at {DASHBOARD_TEMPLATE_FILE}") + # Create a basic HTML structure if template is missing, to show some output + output_content = f""" + +Note: dashboard.html template was not found. This is a fallback display.
+ + + """ + else: + with open(DASHBOARD_TEMPLATE_FILE, "r", encoding="utf-8") as f: + template_content = f.read() + + output_content = template_content.replace("", project_name_header) + output_content = output_content.replace("", stats_overview_html) + output_content = output_content.replace("", secondary_stats_html) + output_content = output_content.replace("", latest_interaction_display_html) + output_content = output_content.replace("", history_entries_html) + + # Check if placeholders were correctly replaced (optional, for debugging) + # if "" in output_content and "" not in stats_overview_html: + # print("Warning: Stats overview placeholder was not replaced.") + # if "" in output_content and "" not in secondary_stats_html: + # print("Warning: Secondary stats placeholder was not replaced.") + # if "" in output_content and "" not in latest_interaction_display_html: + # print("Warning: Latest interaction placeholder was not replaced.") + # if "" in output_content and "" not in history_entries_html: + # print("Warning: History entries placeholder was not replaced.") + + + with open(DASHBOARD_OUTPUT_FILE, "w", encoding="utf-8") as f: + f.write(output_content) + + print(f"Dashboard generated: {DASHBOARD_OUTPUT_FILE.resolve().as_uri()}") + webbrowser.open(DASHBOARD_OUTPUT_FILE.resolve().as_uri()) + + +if __name__ == "__main__": + main() diff --git a/local_analytics/local_analytics_collector.py b/local_analytics/local_analytics_collector.py new file mode 100644 index 000000000..92ee29615 --- /dev/null +++ b/local_analytics/local_analytics_collector.py @@ -0,0 +1,351 @@ +# aider/local_analytics_collector.py +import atexit +import datetime +import logging +import os +import platform +import shelve +import sys +import time +import uuid +import json # Import json module +import re # Import re module + +import litellm + +# Import from the local_analytics package (assuming project_root/local_analytics/dashboard_generator.py) +from local_analytics.dashboard_generator import main + +try: + from aider import __version__ as aider_version_val +except ImportError: + aider_version_val = "unknown" + +# Path constants relative to the project root where Aider is run +DATA_SHELVE_FILE = "local_analytics/aider_analytics_data.shelve" +# Constant for the dashboard HTML file +# REMOVED: DASHBOARD_HTML_FILE = "local_analytics/dashboard.html" +LOG_FILE = "local_analytics/local_analytics_collector.logs" +SESSION_JSONL_FILE = "local_analytics/session.jsonl" # Define the new JSONL file path + +class LocalAnalyticsCollector: + """ + Collects local analytics data for Aider sessions and interactions. + + This class tracks various metrics related to LLM calls, token usage, + code modifications, and session timings. Data is stored locally using + the `shelve` module. + """ + def __init__(self, io, git_root=None, enabled=True): + """ + Initializes the LocalAnalyticsCollector. + + Args: + io: An InputOutput object for user interaction (currently unused beyond holding a reference). + git_root (str, optional): The root directory of the git project. + Defaults to None, in which case the current working directory is used. + enabled (bool, optional): Whether analytics collection is enabled. Defaults to True. + """ + self.io = io # Retain for the final user-facing message + self.enabled = enabled + if not self.enabled: + return + + if git_root: + self.project_name = os.path.basename(os.path.abspath(git_root)) + base_path = git_root + else: + self.project_name = os.path.basename(os.getcwd()) + base_path = os.getcwd() + + self.data_file = os.path.join(base_path, DATA_SHELVE_FILE) + self.log_file = os.path.join(base_path, LOG_FILE) + # Store the dashboard output file path + # REMOVED: self.dashboard_output_file = os.path.join(base_path, DASHBOARD_HTML_FILE) + # Store the session JSONL file path + self.session_jsonl_file = os.path.join(base_path, SESSION_JSONL_FILE) + + + self.session_id = str(uuid.uuid4()) + self.aider_version = aider_version_val + self.platform_info = platform.platform() + self.python_version = f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}" + + self._current_interaction_data = None + self._interaction_start_time_monotonic = None + + # <<< START LOGGER SETUP + log_dir = os.path.dirname(self.log_file) + if log_dir and not os.path.exists(log_dir): + os.makedirs(log_dir, exist_ok=True) + + self.logger = logging.getLogger(__name__ + ".LocalAnalyticsCollector") # Or just __name__ + self.logger.setLevel(logging.DEBUG) + self.logger.propagate = False # Prevent logs from reaching root logger / console + + # Remove existing handlers to prevent duplication if __init__ is called multiple times + for handler in self.logger.handlers[:]: + self.logger.removeHandler(handler) + handler.close() + + fh = logging.FileHandler(self.log_file, encoding='utf-8') + formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(session_id)s - %(message)s') + fh.setFormatter(formatter) + self.logger.addHandler(fh) + + # Make session_id available to logger formatter + self._log_adapter = logging.LoggerAdapter(self.logger, {'session_id': self.session_id}) + + self._log_adapter.debug(f"--- LocalAnalyticsCollector Initialized ---") + self._log_adapter.debug(f"Project: {self.project_name}") + self._log_adapter.debug(f"Data file: {self.data_file}") + self._log_adapter.debug(f"Log file: {self.log_file}") + self._log_adapter.debug(f"Session JSONL file: {self.session_jsonl_file}") + # <<< END LOGGER SETUP + + data_dir = os.path.dirname(self.data_file) + if data_dir and not os.path.exists(data_dir): + os.makedirs(data_dir, exist_ok=True) + # Ensure directory for dashboard.html and session.jsonl also exists + # REMOVED: output_dir = os.path.dirname(self.dashboard_output_file) # Assuming dashboard and jsonl are in the same dir + output_dir = os.path.dirname(self.session_jsonl_file) # Use session_jsonl_file path + if output_dir and not os.path.exists(output_dir): + os.makedirs(output_dir, exist_ok=True) + + + atexit.register(self.end_session) + + self._original_success_callbacks = litellm.success_callback[:] + self._original_failure_callbacks = litellm.failure_callback[:] + if self._litellm_success_callback not in litellm.success_callback: + litellm.success_callback.append(self._litellm_success_callback) + + + + + def start_interaction(self, query, modified_files_in_chat=None): + """ + Starts tracking a new interaction. + + If a previous interaction was in progress, it will be ended first. + + Args: + query (str): The user's query for this interaction. + modified_files_in_chat (list, optional): A list of files modified in the chat context. + Defaults to None. + """ + if not self.enabled: + return + if self._current_interaction_data: + self.end_interaction() # End previous interaction if any + + self._interaction_start_time_monotonic = time.monotonic() + self._current_interaction_data = { + "session_id": self.session_id, + "project_name": self.project_name, + "interaction_timestamp": datetime.datetime.now().isoformat(), + "interaction_duration_seconds": 0, + "query": re.split(r"```diff", query, 1)[0].strip(), + "aider_version": self.aider_version, + "platform_info": self.platform_info, + "python_version": self.python_version, + "token_summary": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0, "estimated_cost": 0.0}, + "models_used_summary": [], + "llm_calls_details": [], + "modified_files_in_chat": modified_files_in_chat or [], + "commits_made_this_interaction": [] + } + + + def end_interaction(self): + """ + Ends the current interaction and saves its data. + + Calculates interaction duration, summarizes model usage, and persists + the interaction data to the shelve database. + """ + if not self.enabled or not self._current_interaction_data: + return + + if self._interaction_start_time_monotonic: + duration = time.monotonic() - self._interaction_start_time_monotonic + self._current_interaction_data["interaction_duration_seconds"] = duration + + # Summarize model usage from detailed calls + model_summary_map = {} + for call in self._current_interaction_data.get("llm_calls_details", []): + model_name = call.get("model", "unknown_model") + entry = model_summary_map.setdefault( + model_name, + { + "name": model_name, + "calls": 0, + "cost": 0.0, + "prompt_tokens": 0, + "completion_tokens": 0, + }, + ) + entry["calls"] += 1 + entry["cost"] += call.get("cost", 0.0) + entry["prompt_tokens"] += call.get("prompt_tokens", 0) + entry["completion_tokens"] += call.get("completion_tokens", 0) + self._current_interaction_data["models_used_summary"] = list(model_summary_map.values()) + + try: + with shelve.open(self.data_file) as db: + interactions = db.get("interactions", []) + interactions.append(self._current_interaction_data) + db["interactions"] = interactions + except Exception as e: + self._log_adapter.error(f"Error saving interaction to shelve: {e}") + + self._current_interaction_data = None + self._interaction_start_time_monotonic = None + + + + + def _litellm_success_callback(self, kwargs, completion_response, start_time, end_time): + """ + Callback for successful LiteLLM calls. + + This method is registered with LiteLLM to capture details of each + successful LLM API call, including token usage and cost. + + Args: + kwargs: Keyword arguments passed to the LiteLLM completion call. + completion_response: The response object from LiteLLM. + start_time: Timestamp when the LLM call started. + end_time: Timestamp when the LLM call ended. + """ + if not self.enabled or not self._current_interaction_data: + return + + model_name = kwargs.get("model", "unknown_model") + usage = getattr(completion_response, "usage", None) + prompt_tokens = getattr(usage, 'prompt_tokens', 0) if usage else 0 + completion_tokens = getattr(usage, 'completion_tokens', 0) if usage else 0 + + cost = 0.0 + try: + # Ensure cost is float, handle potential errors from litellm.completion_cost + calculated_cost = litellm.completion_cost(completion_response=completion_response) + cost = float(calculated_cost) if calculated_cost is not None else 0.0 + except Exception as e: # Broad exception catch if litellm.completion_cost fails + self._log_adapter.warning( + f"Analytics: Could not calculate cost for LLM call. Error: {e}" + ) + cost = 0.0 # Ensure cost is always a float, defaulting to 0.0 on error + + call_detail = { + "model": model_name, + "id": getattr(completion_response, "id", None), + "finish_reason": ( + getattr(completion_response.choices[0], "finish_reason", None) + if hasattr(completion_response, "choices") and completion_response.choices + else None + ), + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "cost": cost, + "timestamp": start_time.isoformat(), + } + self._current_interaction_data["llm_calls_details"].append(call_detail) + + ts = self._current_interaction_data["token_summary"] + ts["prompt_tokens"] += prompt_tokens + ts["completion_tokens"] += completion_tokens + ts["total_tokens"] += prompt_tokens + completion_tokens + ts["estimated_cost"] += cost + + + + + def log_commit(self, commit_hash, commit_message): + """ + Logs a git commit made during the current interaction. + + Args: + commit_hash (str): The hash of the commit. + commit_message (str): The commit message. + """ + if not self.enabled or not self._current_interaction_data: + return + commit_info = {"hash": commit_hash, "message": commit_message} + self._current_interaction_data["commits_made_this_interaction"].append(commit_info) + + + + + def end_session(self): + """ + Ends the analytics collection session. + + Ensures any ongoing interaction is ended, generates the HTML dashboard, + unregisters the atexit handler, and restores original LiteLLM callbacks. + """ + if not self.enabled: # If analytics was never enabled or session already ended. + # Unregister atexit handler early if it was somehow registered without enabling + # This path should ideally not be hit if __init__ logic is correct. + try: + atexit.unregister(self.end_session) + except TypeError: # pragma: no cover + pass # Handler was not registered or other issue + return + + + # End any ongoing interaction first + if self._current_interaction_data: + self.end_interaction() + + # Write all the `shelve` data to session.jsonl + if hasattr(self, 'data_file') and hasattr(self, 'session_jsonl_file'): + try: + with shelve.open(self.data_file, 'r') as db: + interactions = db.get("interactions", []) + + with open(self.session_jsonl_file, 'w', encoding='utf-8') as f: + for interaction in interactions: + # Ensure data is JSON serializable (e.g., handle datetime objects if any slipped through) + # Although datetime is converted to isoformat already, this is a good practice. + # Simple approach: convert to string if not serializable, or use a custom encoder. + # For now, assuming isoformat is sufficient based on start_interaction. + json_line = json.dumps(interaction) + f.write(json_line + '\n') + + # generate dashboard + main() + + if hasattr(self, '_log_adapter'): + self._log_adapter.info(f"Shelve data written to {self.session_jsonl_file}") + + except Exception as e: + if hasattr(self, '_log_adapter'): + self._log_adapter.error(f"Error writing shelve data to JSONL: {e}") + else: # pragma: no cover + print(f"Error writing shelve data to JSONL: {e}") # Fallback if logger not set + + + # Cleanup atexit handler + try: + atexit.unregister(self.end_session) + except TypeError: # pragma: no cover + pass # Handler was not registered or other issue + + # Restore LiteLLM callbacks + # Check if _original_success_callbacks exists before assigning + if hasattr(self, '_original_success_callbacks'): + litellm.success_callback = self._original_success_callbacks + # if hasattr(self, '_original_failure_callbacks'): # If failure callbacks were also stored + # litellm.failure_callback = self._original_failure_callbacks + + if hasattr(self, '_log_adapter'): + self._log_adapter.info("LocalAnalyticsCollector session ended.") + + # Ensure logger handlers are closed to release file locks, especially on Windows + if hasattr(self, 'logger'): # Check if logger was initialized + for handler in self.logger.handlers[:]: + handler.close() + self.logger.removeHandler(handler) + # Set self.enabled to False after cleanup to prevent re-entry or further use + self.enabled = False diff --git a/local_analytics/run_aider_ask_reflection_error.py b/local_analytics/run_aider_ask_reflection_error.py new file mode 100644 index 000000000..0b46f9f68 --- /dev/null +++ b/local_analytics/run_aider_ask_reflection_error.py @@ -0,0 +1,112 @@ +#!/usr/bin/env python + +import pexpect +import sys +import os +import time + +# Define the command to run Aider +# Ensure the path to the .env file is correct for your environment +# This script assumes it's run from a location where 'python -m aider' works +# and the path '~/Dev/aider/.env' is valid. +aider_command = "python -m aider --env-file ~/Dev/aider/.env" + +# Define the expected prompts using regex +# r'> ' matches the main aider prompt +# r'Apply edits\? \[y/n/a/e\] ' matches the edit confirmation prompt +main_prompt = r'> ' +edit_prompt = r'Apply edits\? \[y/n/a/e\] ' + +# Set a timeout for pexpect operations (in seconds) +# Adjust this if your LLM responses are very long or system is slow +timeout_seconds = 300 # 5 minutes + +print(f"Running command: {aider_command}") + +child = None +try: + # Spawn the aider process + # encoding='utf-8' ensures consistent text handling + # timeout sets a default timeout for expect operations + child = pexpect.spawn(aider_command, encoding='utf-8', timeout=timeout_seconds) + + # Optional: Uncomment the line below to see the raw output from the child process + # child.logfile_read = sys.stdout + + # Wait for the initial Aider prompt + print("Waiting for initial prompt...") + child.expect(main_prompt) + print("Initial prompt received.") + + # Change mode to /ask + print("Sending /ask command...") + child.sendline("/ask") + # Wait for the prompt to confirm mode change + child.expect(main_prompt) + print("Mode changed to /ask.") + + # Send the query + query = "what is the reflection error" + print(f"Sending query: '{query}'...") + child.sendline(query) + + # Wait for the LLM response to finish and the prompt to reappear. + # This loop also handles potential edit prompts that might appear + # during or after the LLM's response. + print("Waiting for LLM response and handling potential edit prompts...") + while True: + # Wait for either the edit prompt, the main prompt, EOF, or timeout + index = child.expect([edit_prompt, main_prompt, pexpect.EOF, pexpect.TIMEOUT]) + + if index == 0: + # Matched the edit prompt: 'Apply edits? [y/n/a/e] ' + print("Edit prompt received. Sending 'n' to decline...") + child.sendline("n") + # Continue the loop to wait for the next prompt (could be another edit or the main prompt) + elif index == 1: + # Matched the main prompt: '> ' + # This indicates the LLM response is likely finished and no more edit prompts are pending + print("Main prompt received. LLM response finished.") + break # Exit the loop + elif index == 2: + # Matched EOF - the process exited unexpectedly before we sent /exit + print("ERROR: Process exited unexpectedly (EOF).") + print("Output before EOF:") + print(child.before) + break # Exit the loop + elif index == 3: + # Matched TIMEOUT + print(f"ERROR: Timeout occurred ({timeout_seconds} seconds) while waiting for prompt.") + print("Output before timeout:") + print(child.before) + break # Exit the loop + + # Send the /exit command to quit Aider + print("Sending /exit command...") + child.sendline("/exit") + + # Wait for the process to terminate gracefully + print("Waiting for process to exit...") + child.expect(pexpect.EOF) + print("Process exited.") + +except pexpect.exceptions.TIMEOUT as e: + print(f"ERROR: Timeout exception: {e}") + if child: + print("Output before timeout:") + print(child.before) +except pexpect.exceptions.EOF as e: + print(f"ERROR: EOF exception: {e}") + if child: + print("Output before EOF:") + print(child.before) +except Exception as e: + print(f"An unexpected error occurred: {e}") +finally: + # Ensure the child process is terminated if it's still running + if child and child.isalive(): + print("Terminating child process...") + child.close() + print("Child process terminated.") + +print("Script finished.") diff --git a/local_analytics/run_aider_ask_test.py b/local_analytics/run_aider_ask_test.py new file mode 100644 index 000000000..5ed6bc2dc --- /dev/null +++ b/local_analytics/run_aider_ask_test.py @@ -0,0 +1,184 @@ +import subprocess +import time +import os +import sys +import threading +import queue + +# Define the aider command +# Use the full path to the .env file +# Assumes the script is run from the project root or a location where +# ~/Dev/aider/.env is the correct path. +# Using sys.executable ensures the script runs aider with the same python env. +aider_command = [ + sys.executable, + "-m", "aider", + "--env-file", os.path.expanduser("~/Dev/aider/.env") +] + +# Inputs to send to aider +inputs = [ + "/ask", + "what is the reflection error", + "/exit" +] + +# Expected prompts (as bytes, since we read bytes) +# Use strip() because rich might add spaces or other control characters +MAIN_PROMPT = b"> " +EDIT_PROMPT = b"Apply edits? (y/n/commit/diff/quit) " + +def enqueue_output(out, queue): + """Helper function to read output from a stream and put it in a queue.""" + # Read line by line + for line in iter(out.readline, b''): + queue.put(line) + out.close() + +def run_aider_session(): + print(f"[SCRIPT] Starting aider with command: {' '.join(aider_command)}") + + # Start the subprocess + # Use bufsize=1 for line buffering + # universal_newlines=False to read bytes and reliably detect byte prompts + # stderr is also piped as rich often prints to stderr + process = subprocess.Popen( + aider_command, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + bufsize=1, + universal_newlines=False + ) + + # Queues for stdout and stderr + q_stdout = queue.Queue() + q_stderr = queue.Queue() + + # Start threads to read stdout and stderr asynchronously + t_stdout = threading.Thread(target=enqueue_output, args=(process.stdout, q_stdout)) + t_stderr = threading.Thread(target=enqueue_output, args=(process.stderr, q_stderr)) + t_stdout.daemon = True # Thread dies with the main program + t_stderr.daemon = True + t_stdout.start() + t_stderr.start() + + # Give aider a moment to start and print initial messages + time.sleep(3) # Increased initial sleep slightly + + current_input_index = 0 + # State machine: WAITING_FOR_MAIN_PROMPT, WAITING_FOR_RESPONSE + state = "WAITING_FOR_MAIN_PROMPT" + + print(f"[SCRIPT] Initial state: {state}") + + try: + # Continue as long as the process is running OR there is output in the queues + while process.poll() is None or not q_stdout.empty() or not q_stderr.empty(): + try: + # Get a line from stdout queue with a timeout + # A small timeout allows the loop to check process.poll() and stderr queue + line = q_stdout.get(timeout=0.05) + sys.stdout.buffer.write(line) + sys.stdout.buffer.flush() + + # Check for prompts based on the state + if state == "WAITING_FOR_MAIN_PROMPT": + # Check if the line ends with the main prompt bytes (after stripping) + if line.strip().endswith(MAIN_PROMPT.strip()): + print("\n[SCRIPT] Detected main prompt.") + if current_input_index < len(inputs): + command = inputs[current_input_index] + print(f"[SCRIPT] Sending: {command}") + process.stdin.write((command + "\n").encode()) # Encode string to bytes + process.stdin.flush() + current_input_index += 1 + state = "WAITING_FOR_RESPONSE" # After sending input, wait for response/next prompt + print(f"[SCRIPT] State transition: {state}") + else: + # Should not happen if /exit is the last input, but as a safeguard + print("[SCRIPT] No more inputs defined, waiting for process exit.") + state = "SESSION_COMPLETE" + print(f"[SCRIPT] State transition: {state}") + + elif state == "WAITING_FOR_RESPONSE": + # While waiting for response, we might see an edit prompt or the main prompt + if line.strip().endswith(EDIT_PROMPT.strip()): + print("\n[SCRIPT] Detected edit prompt.") + print("[SCRIPT] Sending: n") + process.stdin.write(b"n\n") # Send 'n' to decline edits + process.stdin.flush() + # Stay in WAITING_FOR_RESPONSE state, as declining might lead to another prompt + print(f"[SCRIPT] State remains: {state}") + elif line.strip().endswith(MAIN_PROMPT.strip()): + print("\n[SCRIPT] Detected main prompt (while waiting for response).") + # Response finished, now ready for next main input + state = "WAITING_FOR_MAIN_PROMPT" + print(f"[SCRIPT] State transition: {state}") + + except queue.Empty: + # No output from stdout, check stderr queue + try: + err_line = q_stderr.get(timeout=0.01) + sys.stderr.buffer.write(err_line) + sys.stderr.buffer.flush() + except queue.Empty: + # No output from either queue, check if process is still running + if process.poll() is not None: + # Process exited and queues are empty, we are done + print("[SCRIPT] Process exited and queues are empty.") + break + # If process is still running but no output, just continue loop and wait + + # Add a small sleep to prevent tight loop if process is slow to produce output + time.sleep(0.01) + + # End of while loop: process exited and queues are drained + + except Exception as e: + print(f"[SCRIPT] An error occurred: {e}") + # Attempt to read remaining output before terminating + try: + # Give threads a moment to finish putting data in queues + t_stdout.join(timeout=1) + t_stderr.join(timeout=1) + # Drain queues + while not q_stdout.empty(): + sys.stdout.buffer.write(q_stdout.get_nowait()) + sys.stdout.buffer.flush() + while not q_stderr.empty(): + sys.stderr.buffer.write(q_stderr.get_nowait()) + sys.stdout.buffer.flush() + except Exception as e_drain: + print(f"[SCRIPT] Error draining queues: {e_drain}") + + + if process.poll() is None: + print("[SCRIPT] Terminating process...") + process.terminate() # Ensure process is terminated on error + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + print("[SCRIPT] Process did not terminate, killing...") + process.kill() + process.wait() + + + finally: + # Ensure process is waited upon if not already + if process.poll() is None: + print("[SCRIPT] Waiting for process to finish...") + process.wait() + + # Final drain of queues just in case + while not q_stdout.empty(): + sys.stdout.buffer.write(q_stdout.get_nowait()) + sys.stdout.buffer.flush() + while not q_stderr.empty(): + sys.stderr.buffer.write(q_stderr.get_nowait()) + sys.stderr.buffer.flush() + + print(f"[SCRIPT] Aider process finished with return code {process.returncode}") + +if __name__ == "__main__": + run_aider_session() diff --git a/tests/local_analytics/test_local_analytics_collector_standalone.py b/tests/local_analytics/test_local_analytics_collector_standalone.py new file mode 100644 index 000000000..1f6384e3e --- /dev/null +++ b/tests/local_analytics/test_local_analytics_collector_standalone.py @@ -0,0 +1,230 @@ +import os +import shelve +import json +import tempfile +import shutil +import unittest +from unittest.mock import patch, MagicMock +import datetime +import time +import logging # Import logging for logger checks + +# Assuming the script is run from the project root or PYTHONPATH is set +# This import path assumes the test script is in tests/local_analytics/ +try: + from local_analytics.local_analytics_collector import LocalAnalyticsCollector +except ImportError: + # Fallback import path if the script is run from a different location + # This might require adjusting PYTHONPATH or running from the project root + print("Could not import LocalAnalyticsCollector directly. Ensure PYTHONPATH is set or run from project root.") + print("Attempting import assuming script is in tests/local_analytics/") + try: + # Adjust path for potential different execution contexts + import sys + sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..'))) + from local_analytics.local_analytics_collector import LocalAnalyticsCollector + sys.path.pop(0) # Clean up sys.path + except ImportError as e: + print(f"Failed to import LocalAnalyticsCollector even with path adjustment: {e}") + # Exit or raise error if import fails + raise + + +# Dummy IO class to satisfy the collector's __init__ +class DummyIO: + def tool_output(self, *args, **kwargs): + pass + def tool_warning(self, *args, **kwargs): + pass + def tool_error(self, *args, **kwargs): + pass + def confirm_ask(self, *args, **kwargs): + return 'y' # Default to yes for confirmations + def print(self, *args, **kwargs): + pass + def append_chat_history(self, *args, **kwargs): + pass + + +class TestLocalAnalyticsCollectorStandalone(unittest.TestCase): + + def setUp(self): + # Create a temporary directory for test files + self.temp_dir = tempfile.mkdtemp() + self.project_name = os.path.basename(self.temp_dir) + # Define file paths relative to the temporary project root + self.analytics_dir = os.path.join(self.temp_dir, "local_analytics") + self.data_file = os.path.join(self.analytics_dir, "aider_analytics_data.shelve") + self.session_jsonl_file = os.path.join(self.analytics_dir, "session.jsonl") + self.dashboard_output_file = os.path.join(self.analytics_dir, "dashboard.html") + self.log_file = os.path.join(self.analytics_dir, "local_analytics_collector.logs") + + # Ensure the local_analytics directory exists within the temp dir + os.makedirs(self.analytics_dir, exist_ok=True) + + # Mock the generate_dashboard function + # Patch the function where it's *used* in local_analytics_collector.py + self.patcher_generate_dashboard = patch('local_analytics.local_analytics_collector.generate_dashboard') + self.mock_generate_dashboard = self.patcher_generate_dashboard.start() + + # Mock litellm.completion_cost as it might be called internally + self.patcher_litellm_cost = patch('litellm.completion_cost') + self.mock_litellm_cost = self.patcher_litellm_cost.start() + self.mock_litellm_cost.return_value = 0.03 # Return a fixed cost for testing + + # Mock litellm.success_callback list to control it during the test + # The collector appends its callback to this list in __init__ + self.patcher_litellm_success_callback_list = patch('litellm.success_callback', new_callable=list) + self.mock_litellm_success_callback_list = self.patcher_litellm_success_callback_list.start() + + # Create a dummy IO object + self.dummy_io = DummyIO() + + def tearDown(self): + # Stop all patches + self.patcher_generate_dashboard.stop() + self.patcher_litellm_cost.stop() + self.patcher_litellm_success_callback_list.stop() + + # Clean up the temporary directory + if os.path.exists(self.temp_dir): + shutil.rmtree(self.temp_dir) + + def test_analytics_collection_and_output(self): + """ + Tests that analytics data is collected, saved to shelve, + written to session.jsonl, and the dashboard generator is called + with the correct shelve file path. + """ + # 1. Initialize collector + # Pass the temporary directory as the git_root + collector = LocalAnalyticsCollector(self.dummy_io, git_root=self.temp_dir, enabled=True) + + # Verify the collector's callback was added to the litellm list + self.assertIn(collector._litellm_success_callback, self.mock_litellm_success_callback_list) + + # 2. Simulate an interaction + query = "Test query for analytics collection" + modified_files = ["test_file1.py", "docs/test_doc.md"] + collector.start_interaction(query, modified_files_in_chat=modified_files) + + # Simulate an LLM call within the interaction + mock_completion_response = MagicMock() + mock_completion_response.usage.prompt_tokens = 100 + mock_completion_response.usage.completion_tokens = 200 + mock_completion_response.id = "chatcmpl-test-id-12345" + mock_completion_response.choices = [MagicMock()] + mock_completion_response.choices[0].finish_reason = "stop" + + llm_call_kwargs = {"model": "gpt-4o", "messages": [{"role": "user", "content": "..."}]} + start_time = datetime.datetime.now() + # Simulate some duration + time.sleep(0.01) + end_time = datetime.datetime.now() + + # Manually call the internal success callback to simulate a completed LLM call + collector._litellm_success_callback(llm_call_kwargs, mock_completion_response, start_time, end_time) + + # Simulate a commit + commit_hash = "abcdef1234567890" + commit_message = "feat: added test analytics data" + collector.log_commit(commit_hash, commit_message) + + # End the interaction + collector.end_interaction() + + # 3. End the session (triggers saving to shelve and writing to jsonl) + collector.end_session() + + # 4. Assertions + + # Check if shelve file exists and contains data + # Shelve creates multiple files, check for the base name + self.assertTrue(any(f.startswith(os.path.basename(self.data_file)) for f in os.listdir(self.analytics_dir)), + "Shelve data files should exist") + try: + # Use the base path for shelve.open + with shelve.open(self.data_file, 'r') as db: + self.assertIn("interactions", db, "Shelve should contain 'interactions' key") + interactions = db["interactions"] + self.assertIsInstance(interactions, list, "'interactions' in shelve should be a list") + self.assertEqual(len(interactions), 1, "Shelve should contain exactly one interaction") + + interaction_data = interactions[0] + self.assertEqual(interaction_data.get("query"), query) + self.assertEqual(interaction_data.get("modified_files_in_chat"), modified_files) + self.assertGreater(interaction_data.get("interaction_duration_seconds", 0), 0) + + self.assertIn("llm_calls_details", interaction_data) + self.assertEqual(len(interaction_data["llm_calls_details"]), 1) + llm_call_detail = interaction_data["llm_calls_details"][0] + self.assertEqual(llm_call_detail.get("model"), "gpt-4o") + self.assertEqual(llm_call_detail.get("prompt_tokens"), 100) + self.assertEqual(llm_call_detail.get("completion_tokens"), 200) + self.assertEqual(llm_call_detail.get("cost"), 0.03) + # Check timestamp format (isoformat) + self.assertIsInstance(llm_call_detail.get("timestamp"), str) + try: + datetime.datetime.fromisoformat(llm_call_detail["timestamp"]) + except ValueError: + self.fail("LLM call timestamp is not in ISO format") + + + self.assertIn("commits_made_this_interaction", interaction_data) + self.assertEqual(len(interaction_data["commits_made_this_interaction"]), 1) + self.assertEqual(interaction_data["commits_made_this_interaction"][0].get("hash"), commit_hash) + self.assertEqual(interaction_data["commits_made_this_interaction"][0].get("message"), commit_message) + + # Check token summary + token_summary = interaction_data.get("token_summary", {}) + self.assertEqual(token_summary.get("prompt_tokens"), 100) + self.assertEqual(token_summary.get("completion_tokens"), 200) + self.assertEqual(token_summary.get("total_tokens"), 300) + self.assertEqual(token_summary.get("estimated_cost"), 0.03) + + + except Exception as e: + self.fail(f"Error reading shelve file: {e}") + + # Check if session.jsonl file exists and contains data + self.assertTrue(os.path.exists(self.session_jsonl_file), "session.jsonl file should exist") + try: + with open(self.session_jsonl_file, 'r', encoding='utf-8') as f: + lines = f.readlines() + self.assertEqual(len(lines), 1, "session.jsonl should contain exactly one line") + json_data = json.loads(lines[0]) + + # Verify content matches the interaction data saved in shelve + # Note: JSON serialization/deserialization might change types slightly (e.g., datetime becomes string) + # We already verified the shelve data structure above, just check some key values + self.assertIsInstance(json_data, dict) + self.assertEqual(json_data.get("query"), query) + self.assertEqual(json_data.get("modified_files_in_chat"), modified_files) + self.assertIn("llm_calls_details", json_data) + self.assertEqual(len(json_data["llm_calls_details"]), 1) + self.assertIn("commits_made_this_interaction", json_data) + self.assertEqual(len(json_data["commits_made_this_interaction"]), 1) + self.assertEqual(json_data.get("token_summary", {}).get("total_tokens"), 300) + + + except Exception as e: + self.fail(f"Error reading or parsing session.jsonl: {e}") + + + # Check if generate_dashboard was called with correct arguments + self.mock_generate_dashboard.assert_called_once() + # Check arguments: project_name, shelve_file_path, dashboard_output_path, logger + called_args, called_kwargs = self.mock_generate_dashboard.call_args + self.assertEqual(called_args[0], self.project_name) + self.assertEqual(called_args[1], self.data_file) # Verify shelve file path is passed + self.assertEqual(called_args[2], self.dashboard_output_file) + # Optionally check the logger argument type + self.assertIsInstance(called_args[3], logging.LoggerAdapter) + + +# This allows running the test directly from the command line +if __name__ == '__main__': + # Add a basic handler for unittest output if run directly + logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') + unittest.main(argv=['first-arg-is-ignored'], exit=False) +