From 8c5a560787fac946a358267a71cec48080ba134a Mon Sep 17 00:00:00 2001 From: Ed_ Date: Sat, 28 Feb 2026 18:35:54 -0500 Subject: [PATCH] refactor(ai_client): Add strict type hints to global variables --- ai_client.py | 54 ++++++++++++++++++++++++++-------------------------- 1 file changed, 27 insertions(+), 27 deletions(-) diff --git a/ai_client.py b/ai_client.py index 5734062..dfd529a 100644 --- a/ai_client.py +++ b/ai_client.py @@ -40,7 +40,7 @@ _max_tokens: int = 8192 _history_trunc_limit: int = 8000 # Global event emitter for API lifecycle events -events = EventEmitter() +events: EventEmitter = EventEmitter() def set_model_params(temp: float, max_tok: int, trunc_limit: int = 8000) -> None: global _temperature, _max_tokens, _history_trunc_limit @@ -55,52 +55,52 @@ def set_history_trunc_limit(val: int) -> None: global _history_trunc_limit _history_trunc_limit = val -_gemini_client = None -_gemini_chat = None -_gemini_cache = None +_gemini_client: genai.Client | None = None +_gemini_chat: Any = None +_gemini_cache: Any = None _gemini_cache_md_hash: int | None = None _gemini_cache_created_at: float | None = None # Gemini cache TTL in seconds. Caches are created with this TTL and # proactively rebuilt at 90% of this value to avoid stale-reference errors. -_GEMINI_CACHE_TTL = 3600 +_GEMINI_CACHE_TTL: int = 3600 -_anthropic_client = None +_anthropic_client: anthropic.Anthropic | None = None _anthropic_history: list[dict] = [] -_anthropic_history_lock = threading.Lock() +_anthropic_history_lock: threading.Lock = threading.Lock() -_deepseek_client = None +_deepseek_client: Any = None _deepseek_history: list[dict] = [] -_deepseek_history_lock = threading.Lock() +_deepseek_history_lock: threading.Lock = threading.Lock() -_send_lock = threading.Lock() +_send_lock: threading.Lock = threading.Lock() -_gemini_cli_adapter = None +_gemini_cli_adapter: GeminiCliAdapter | None = None # Injected by gui.py - called when AI wants to run a command. # Signature: (script: str, base_dir: str) -> str | None -confirm_and_run_callback = None +confirm_and_run_callback: Callable[[str, str], str | None] | None = None # Injected by gui.py - called whenever a comms entry is appended. # Signature: (entry: dict) -> None -comms_log_callback = None +comms_log_callback: Callable[[dict[str, Any]], None] | None = None # Injected by gui.py - called whenever a tool call completes. # Signature: (script: str, result: str) -> None -tool_log_callback = None +tool_log_callback: Callable[[str, str], None] | None = None # Increased to allow thorough code exploration before forcing a summary -MAX_TOOL_ROUNDS = 10 +MAX_TOOL_ROUNDS: int = 10 # Maximum cumulative bytes of tool output allowed per send() call. # Prevents unbounded memory growth during long tool-calling loops. -_MAX_TOOL_OUTPUT_BYTES = 500_000 +_MAX_TOOL_OUTPUT_BYTES: int = 500_000 # Maximum characters per text chunk sent to Anthropic. # Kept well under the ~200k token API limit. -_ANTHROPIC_CHUNK_SIZE = 120_000 +_ANTHROPIC_CHUNK_SIZE: int = 120_000 -_SYSTEM_PROMPT = ( +_SYSTEM_PROMPT: str = ( "You are a helpful coding assistant with access to a PowerShell tool and MCP tools (file access: read_file, list_directory, search_files, get_file_summary, web access: web_search, fetch_url). " "When calling file/directory tools, always use the 'path' parameter for the target path. " "When asked to create or edit files, prefer targeted edits over full rewrites. " @@ -130,7 +130,7 @@ def _get_combined_system_prompt() -> str: _comms_log: list[dict] = [] -COMMS_CLAMP_CHARS = 300 +COMMS_CLAMP_CHARS: int = 300 def _append_comms(direction: str, kind: str, payload: dict[str, Any]) -> None: entry = { @@ -379,7 +379,7 @@ def _list_deepseek_models(api_key: str) -> list[str]: return ["deepseek-chat", "deepseek-reasoner", "deepseek-v3", "deepseek-r1"] # ------------------------------------------------------------------ tool definition -TOOL_NAME = "run_powershell" +TOOL_NAME: str = "run_powershell" _agent_tools: dict = {} @@ -427,9 +427,9 @@ def _build_anthropic_tools() -> list[dict]: tools_list[-1]["cache_control"] = {"type": "ephemeral"} return tools_list -_ANTHROPIC_TOOLS = _build_anthropic_tools() +_ANTHROPIC_TOOLS: list[dict[str, Any]] = _build_anthropic_tools() -_CACHED_ANTHROPIC_TOOLS = None +_CACHED_ANTHROPIC_TOOLS: list[dict[str, Any]] | None = None def _get_anthropic_tools() -> list[dict]: """Return the Anthropic tools list, rebuilding only once per session.""" @@ -550,7 +550,7 @@ def _build_file_context_text(file_items: list[dict]) -> str: parts.append(f"### `{path}`\n\n```{suffix}\n{content}\n```") return "\n\n---\n\n".join(parts) -_DIFF_LINE_THRESHOLD = 200 +_DIFF_LINE_THRESHOLD: int = 200 def _build_file_diff_text(changed_items: list[dict]) -> str: """ @@ -933,18 +933,18 @@ def _send_gemini_cli(md_content: str, user_message: str, base_dir: str, # ------------------------------------------------------------------ anthropic history management # Rough chars-per-token ratio. Anthropic tokeniser averages ~3.5-4 chars/token. # We use 3.5 to be conservative (overestimate token count = safer). -_CHARS_PER_TOKEN = 3.5 +_CHARS_PER_TOKEN: float = 3.5 # Maximum token budget for the entire prompt (system + tools + messages). # Anthropic's limit is 200k. We leave headroom for the response + tool schemas. -_ANTHROPIC_MAX_PROMPT_TOKENS = 180_000 +_ANTHROPIC_MAX_PROMPT_TOKENS: int = 180_000 # Gemini models have a 1M context window but we cap well below to leave headroom. # If the model reports input tokens exceeding this, we trim old history. -_GEMINI_MAX_INPUT_TOKENS = 900_000 +_GEMINI_MAX_INPUT_TOKENS: int = 900_000 # Marker prefix used to identify stale file-refresh injections in history -_FILE_REFRESH_MARKER = "[FILES UPDATED" +_FILE_REFRESH_MARKER: str = "[FILES UPDATED" def _estimate_message_tokens(msg: dict) -> int: """