refactor(ai_client): Add strict type hints to global variables

This commit is contained in:
2026-02-28 18:35:54 -05:00
parent 42af2e1fa4
commit 8c5a560787

View File

@@ -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:
"""