diff --git a/src/ai_client.py b/src/ai_client.py index b8f0be85..f415069c 100644 --- a/src/ai_client.py +++ b/src/ai_client.py @@ -62,6 +62,18 @@ PROVIDERS: List[str] = ["gemini", "anthropic", "gemini_cli", "deepseek", "minima # hasattr(src.ai_client, '_require_warmed')) continue to work. from src.module_loader import _require_warmed # noqa: E402,F401 from src.result_types import ErrorInfo, ErrorKind, Result # noqa: E402,F401 +from src.type_aliases import ( + CommsLog, + CommsLogCallback, + CommsLogEntry, + FileItem, + FileItems, + History, + HistoryMessage, + Metadata, + ToolCall, + ToolDefinition, +) _provider: str = "gemini" _model: str = "gemini-2.5-flash-lite" @@ -96,28 +108,28 @@ _gemini_cached_file_paths: list[str] = [] _GEMINI_CACHE_TTL: int = 3600 _anthropic_client: Optional[anthropic.Anthropic] = None -_anthropic_history: list[dict[str, Any]] = [] +_anthropic_history: list[Metadata] = [] _anthropic_history_lock: threading.Lock = threading.Lock() _deepseek_client: Any = None -_deepseek_history: list[dict[str, Any]] = [] +_deepseek_history: list[Metadata] = [] _deepseek_history_lock: threading.Lock = threading.Lock() _minimax_client: Any = None -_minimax_history: list[dict[str, Any]] = [] +_minimax_history: list[Metadata] = [] _minimax_history_lock: threading.Lock = threading.Lock() _qwen_client: Any = None -_qwen_history: list[dict[str, Any]] = [] +_qwen_history: list[Metadata] = [] _qwen_history_lock: threading.Lock = threading.Lock() _qwen_region: str = "china" _grok_client: Any = None -_grok_history: list[dict[str, Any]] = [] +_grok_history: list[Metadata] = [] _grok_history_lock: threading.Lock = threading.Lock() _llama_client: Any = None -_llama_history: list[dict[str, Any]] = [] +_llama_history: list[Metadata] = [] _llama_history_lock: threading.Lock = threading.Lock() _llama_base_url: str = "http://localhost:11434/v1" _llama_api_key: str = "ollama" @@ -135,7 +147,7 @@ confirm_and_run_callback: Optional[Callable[[str, str, Optional[Callable[[str], # Injected by gui.py - called whenever a comms entry is appended. # Use get_comms_log_callback/set_comms_log_callback for thread-safe access. -comms_log_callback: Optional[Callable[[dict[str, Any]], None]] = None +comms_log_callback: Optional[CommsLogCallback] = None # Injected by gui.py - called whenever a tool call completes. tool_log_callback: Optional[Callable[[str, str], None]] = None @@ -224,7 +236,7 @@ def _get_combined_system_prompt(preset: Optional[ToolPreset] = None, bias: Optio def get_combined_system_prompt(preset: Optional[ToolPreset] = None, bias: Optional[BiasProfile] = None) -> str: return _get_combined_system_prompt(preset, bias) -_comms_log: deque[dict[str, Any]] = deque(maxlen=1000) +_comms_log: deque[CommsLogEntry] = deque(maxlen=1000) COMMS_CLAMP_CHARS: int = 300 @@ -232,18 +244,18 @@ COMMS_CLAMP_CHARS: int = 300 #region: Comms Log -def get_comms_log_callback() -> Optional[Callable[[dict[str, Any]], None]]: +def get_comms_log_callback() -> Optional[CommsLogCallback]: tl_cb = getattr(_local_storage, "comms_log_callback", None) if tl_cb: return tl_cb return comms_log_callback -def set_comms_log_callback(cb: Optional[Callable[[dict[str, Any]], None]]) -> None: +def set_comms_log_callback(cb: Optional[CommsLogCallback]) -> None: global comms_log_callback comms_log_callback = cb _local_storage.comms_log_callback = cb -def _append_comms(direction: str, kind: str, payload: dict[str, Any]) -> None: - entry: dict[str, Any] = { +def _append_comms(direction: str, kind: str, payload: Metadata) -> None: + entry: Metadata = { "ts": datetime.datetime.now().strftime("%H:%M:%S"), "direction": direction, "kind": kind, @@ -258,7 +270,7 @@ def _append_comms(direction: str, kind: str, payload: dict[str, Any]) -> None: if _cb is not None: _cb(entry) -def get_comms_log() -> list[dict[str, Any]]: +def get_comms_log() -> list[Metadata]: return list(_comms_log) def clear_comms_log() -> None: @@ -267,7 +279,7 @@ def clear_comms_log() -> None: def get_credentials_path() -> Path: return Path(os.environ.get("SLOP_CREDENTIALS", str(Path(__file__).parent.parent / "credentials.toml"))) -def _load_credentials() -> dict[str, Any]: +def _load_credentials() -> Metadata: cred_path = get_credentials_path() try: with open(cred_path, "rb") as f: @@ -608,11 +620,11 @@ def get_bias_profile() -> Optional[str]: """Returns the name of the currently active bias profile.""" return _active_bias_profile.name if _active_bias_profile else None -def _build_anthropic_tools() -> list[dict[str, Any]]: +def _build_anthropic_tools() -> list[ToolDefinition]: """ [C: tests/test_agent_tools_wiring.py:test_build_anthropic_tools_conversion, tests/test_tool_access_exclusion.py:test_build_anthropic_tools_excludes_disabled] """ - raw_tools: list[dict[str, Any]] = [] + raw_tools: list[Metadata] = [] for spec in mcp_client.get_tool_schemas(): if _agent_tools.get(spec["name"], True): raw_tools.append({ @@ -647,9 +659,9 @@ def _build_anthropic_tools() -> list[dict[str, Any]]: raw_tools[-1]["cache_control"] = {"type": "ephemeral"} return raw_tools -_CACHED_ANTHROPIC_TOOLS: Optional[list[dict[str, Any]]] = None +_CACHED_ANTHROPIC_TOOLS: Optional[FileItems] = None -def _get_anthropic_tools() -> list[dict[str, Any]]: +def _get_anthropic_tools() -> list[Metadata]: """ [C: tests/test_bias_efficacy.py:test_bias_efficacy_prompt_generation, tests/test_bias_efficacy.py:test_bias_parameter_nudging, tests/test_bias_integration.py:test_tool_declaration_biasing_anthropic] """ @@ -669,7 +681,7 @@ def _gemini_tool_declaration() -> Optional[types.Tool]: # completes the chain once, then `.types` is just an attribute access. genai = _require_warmed("google.genai") types = genai.types - raw_tools: list[dict[str, Any]] = [] + raw_tools: list[Metadata] = [] for spec in mcp_client.get_tool_schemas(): if _agent_tools.get(spec["name"], True): raw_tools.append({ @@ -726,7 +738,7 @@ def _gemini_tool_declaration() -> Optional[types.Tool]: #region: Tool Execution -def _parse_tool_args_result(tool_args_str: str) -> Result[dict[str, Any]]: +def _parse_tool_args_result(tool_args_str: str) -> Result[Metadata]: """Parse tool call arguments from JSON. Returns Result[dict, ErrorInfo]. On JSON parse failure, returns Result(data={}, errors=[ErrorInfo(...)]). @@ -789,8 +801,8 @@ async def _execute_tool_calls_concurrently( tasks = [] for fc in calls: if provider == "gemini": name, args, call_id = fc.name, dict(fc.args), fc.name # Gemini 1.0.0 doesn't have call IDs in types.Part - elif provider == "gemini_cli": name, args, call_id = cast(str, fc.get("name")), cast(dict[str, Any], fc.get("args", {})), cast(str, fc.get("id")) - elif provider == "anthropic": name, args, call_id = cast(str, getattr(fc, "name")), cast(dict[str, Any], getattr(fc, "input")), cast(str, getattr(fc, "id")) + elif provider == "gemini_cli": name, args, call_id = cast(str, fc.get("name")), cast(Metadata, fc.get("args", {})), cast(str, fc.get("id")) + elif provider == "anthropic": name, args, call_id = cast(str, getattr(fc, "name")), cast(Metadata, getattr(fc, "input")), cast(str, getattr(fc, "id")) elif provider == "deepseek": tool_info = fc.get("function", {}) name = cast(str, tool_info.get("name")) @@ -830,11 +842,11 @@ def run_with_tool_loop( base_dir: str, vendor_name: str, history_lock: Optional[threading.Lock] = None, - history: Optional[list[dict[str, Any]]] = None, - trim_func: Optional[Callable[[list[dict[str, Any]]], None]] = None, + history: Optional[FileItems] = None, + trim_func: Optional[Callable[[list[Metadata]], None]] = None, reasoning_extractor: Optional[Callable[[Any], str]] = None, send_func: Optional[Callable[[int], NormalizedResponse]] = None, - on_pre_dispatch: Optional[Callable[[int, list[dict[str, Any]]], list[dict[str, Any]]]] = None, + on_pre_dispatch: Optional[Callable[[int, list[Metadata]], list[Metadata]]] = None, wrap_reasoning_in_text: bool = False, ) -> str: """ @@ -855,7 +867,7 @@ def run_with_tool_loop( base_dir (str): Base workspace directory. vendor_name (str): The vendor name. history_lock (Optional[threading.Lock]): Lock for thread safety on history. - history (Optional[list[dict[str, Any]]]): Conversation history. + history (Optional[FileItems]): Conversation history. trim_func (Optional[Callable]): Trimming callback for history. reasoning_extractor (Optional[Callable]): Callback to extract reasoning content. send_func (Optional[Callable]): Dispatch sender callback. @@ -898,7 +910,7 @@ def run_with_tool_loop( response_text = response.text or "" if history_lock is not None and history is not None: with history_lock: - msg: dict[str, Any] = {"role": "assistant", "content": response.text or None} + msg: Metadata = {"role": "assistant", "content": response.text or None} if reasoning_content: msg["reasoning_content"] = reasoning_content if response.tool_calls: msg["tool_calls"] = response.tool_calls history.append(msg) @@ -932,7 +944,7 @@ def run_with_tool_loop( async def _execute_single_tool_call_async( name: str, - args: dict[str, Any], + args: Metadata, call_id: str, base_dir: str, pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]], @@ -950,7 +962,7 @@ async def _execute_single_tool_call_async( Parameters & Inputs: name (str): The name of the tool to execute. - args (dict[str, Any]): Arguments passed to the tool. + args (Metadata): Arguments passed to the tool. call_id (str): Unique call identifier. base_dir (str): Workspace root directory. pre_tool_callback (Optional[Callable]): Hook for HITL validation. @@ -1041,7 +1053,7 @@ def _truncate_tool_output(output: str) -> str: #region: File Context Building -def _reread_file_items_result(file_items: list[dict[str, Any]]) -> Result[tuple[list[dict[str, Any]], list[dict[str, Any]]]]: +def _reread_file_items_result(file_items: FileItems) -> Result[FileItemsDiff]: """Re-reads file items, returns (refreshed, changed) tuple. Per-file read errors are accumulated into Result.errors (structured @@ -1050,8 +1062,8 @@ def _reread_file_items_result(file_items: list[dict[str, Any]]) -> Result[tuple[ future callers should check result.errors to detect file re-read failures. """ - refreshed: list[dict[str, Any]] = [] - changed: list[dict[str, Any]] = [] + refreshed: list[Metadata] = [] + changed: list[Metadata] = [] errors: list[ErrorInfo] = [] for item in file_items: path = item.get("path") @@ -1077,7 +1089,7 @@ def _reread_file_items_result(file_items: list[dict[str, Any]]) -> Result[tuple[ return Result(data=(refreshed, changed), errors=errors) -def _build_file_context_text(file_items: list[dict[str, Any]]) -> str: +def _build_file_context_text(file_items: FileItems) -> str: if not file_items: return "" parts: list[str] = [] @@ -1090,7 +1102,7 @@ def _build_file_context_text(file_items: list[dict[str, Any]]) -> str: _DIFF_LINE_THRESHOLD: int = 200 -def _build_file_diff_text(changed_items: list[dict[str, Any]]) -> str: +def _build_file_diff_text(changed_items: FileItems) -> str: """ Generates unified diffs or full file dumps for changed files in the context. @@ -1099,7 +1111,7 @@ def _build_file_diff_text(changed_items: list[dict[str, Any]]) -> str: the full file is dumped; otherwise, a unified diff is constructed. Parameters & Inputs: - changed_items (list[dict[str, Any]]): List of file dictionaries that have changed. + changed_items (list[Metadata]): List of file dictionaries that have changed. Returns: str: Combined markdown string representing the changes or full files. @@ -1133,8 +1145,8 @@ def _build_file_diff_text(changed_items: list[dict[str, Any]]) -> str: else: parts.append(f"### `{path}` (no changes detected)") return "\n\n---\n\n".join(parts) -def _build_deepseek_tools() -> list[dict[str, Any]]: - raw_tools: list[dict[str, Any]] = [] +def _build_deepseek_tools() -> list[ToolDefinition]: + raw_tools: list[Metadata] = [] for spec in mcp_client.get_tool_schemas(): if _agent_tools.get(spec["name"], True): raw_tools.append({ @@ -1165,7 +1177,7 @@ def _build_deepseek_tools() -> list[dict[str, Any]]: }) if _active_tool_preset: _BIAS_ENGINE.apply_semantic_nudges(raw_tools, _active_tool_preset) - tools_list: list[dict[str, Any]] = [] + tools_list: list[Metadata] = [] for tool_def in raw_tools: tools_list.append({ "type": "function", @@ -1177,18 +1189,18 @@ def _build_deepseek_tools() -> list[dict[str, Any]]: }) return tools_list -_CACHED_DEEPSEEK_TOOLS: Optional[list[dict[str, Any]]] = None +_CACHED_DEEPSEEK_TOOLS: Optional[FileItems] = None -def _get_deepseek_tools() -> list[dict[str, Any]]: +def _get_deepseek_tools() -> list[Metadata]: global _CACHED_DEEPSEEK_TOOLS if _CACHED_DEEPSEEK_TOOLS is None: _CACHED_DEEPSEEK_TOOLS = _build_deepseek_tools() return _CACHED_DEEPSEEK_TOOLS -def _content_block_to_dict(block: Any) -> dict[str, Any]: +def _content_block_to_dict(block: Any) -> Metadata: if isinstance(block, dict): return block - if hasattr(block, "model_dump"): return cast(dict[str, Any], block.model_dump()) - if hasattr(block, "to_dict"): return cast(dict[str, Any], block.to_dict()) + if hasattr(block, "model_dump"): return cast(Metadata, block.model_dump()) + if hasattr(block, "to_dict"): return cast(Metadata, block.to_dict()) block_type = getattr(block, "type", None) if block_type == "text": return {"type": "text", "text": block.text} if block_type == "tool_use": return {"type": "tool_use", "id": getattr(block, "id"), "name": getattr(block, "name"), "input": getattr(block, "input")} @@ -1203,7 +1215,7 @@ _ANTHROPIC_MAX_PROMPT_TOKENS: int = 180_000 _GEMINI_MAX_INPUT_TOKENS: int = 900_000 _FILE_REFRESH_MARKER: str = _project_context_marker if _project_context_marker.strip() else "[SYSTEM: FILES UPDATED]" -def _estimate_message_tokens(msg: dict[str, Any]) -> int: +def _estimate_message_tokens(msg: Metadata) -> int: cached = msg.get("_est_tokens") if cached is not None: return cast(int, cached) total_chars = 0 @@ -1225,10 +1237,10 @@ def _estimate_message_tokens(msg: dict[str, Any]) -> int: msg["_est_tokens"] = est return est -def _invalidate_token_estimate(msg: dict[str, Any]) -> None: +def _invalidate_token_estimate(msg: Metadata) -> None: msg.pop("_est_tokens", None) -def _estimate_prompt_tokens(system_blocks: list[dict[str, Any]], history: list[dict[str, Any]]) -> int: +def _estimate_prompt_tokens(system_blocks: list[Metadata], history: list[Metadata]) -> int: total = 0 for block in system_blocks: text = cast(str, block.get("text", "")) @@ -1238,7 +1250,7 @@ def _estimate_prompt_tokens(system_blocks: list[dict[str, Any]], history: list[d total += _estimate_message_tokens(msg) return total -def _strip_stale_file_refreshes(history: list[dict[str, Any]]) -> None: +def _strip_stale_file_refreshes(history: list[Metadata]) -> None: if len(history) < 2: return last_user_idx = -1 @@ -1252,7 +1264,7 @@ def _strip_stale_file_refreshes(history: list[dict[str, Any]]) -> None: content = msg.get("content") if not isinstance(content, list): continue - cleaned: list[dict[str, Any]] = [] + cleaned: list[Metadata] = [] for block in content: if isinstance(block, dict) and block.get("type") == "text": text = cast(str, block.get("text", "")) @@ -1266,17 +1278,17 @@ def _strip_stale_file_refreshes(history: list[dict[str, Any]]) -> None: def _chunk_text(text: str, chunk_size: int) -> list[str]: return [text[i:i + chunk_size] for i in range(0, len(text), chunk_size)] -def _build_chunked_context_blocks(md_content: str) -> list[dict[str, Any]]: +def _build_chunked_context_blocks(md_content: str) -> list[Metadata]: chunks = _chunk_text(md_content, _ANTHROPIC_CHUNK_SIZE) - blocks: list[dict[str, Any]] = [] + blocks: list[Metadata] = [] for i, chunk in enumerate(chunks): - block: dict[str, Any] = {"type": "text", "text": chunk} + block: Metadata = {"type": "text", "text": chunk} if i == len(chunks) - 1: block["cache_control"] = {"type": "ephemeral"} blocks.append(block) return blocks -def _strip_cache_controls(history: list[dict[str, Any]]) -> None: +def _strip_cache_controls(history: list[Metadata]) -> None: for msg in history: content = msg.get("content") if isinstance(content, list): @@ -1284,7 +1296,7 @@ def _strip_cache_controls(history: list[dict[str, Any]]) -> None: if isinstance(block, dict): block.pop("cache_control", None) -def _add_history_cache_breakpoint(history: list[dict[str, Any]]) -> None: +def _add_history_cache_breakpoint(history: list[Metadata]) -> None: user_indices = [i for i, m in enumerate(history) if m.get("role") == "user"] if len(user_indices) < 2: return target_idx = user_indices[-2] @@ -1338,7 +1350,7 @@ def _ensure_anthropic_client() -> None: default_headers = {"anthropic-beta": "prompt-caching-2024-07-31"} ) -def _trim_anthropic_history(system_blocks: list[dict[str, Any]], history: list[dict[str, Any]]) -> int: +def _trim_anthropic_history(system_blocks: list[Metadata], history: list[Metadata]) -> int: _strip_stale_file_refreshes(history) est = _estimate_prompt_tokens(system_blocks, history) if est <= _ANTHROPIC_MAX_PROMPT_TOKENS: return 0 @@ -1366,7 +1378,7 @@ def _trim_anthropic_history(system_blocks: list[dict[str, Any]], history: list[d est -= _estimate_message_tokens(removed) return dropped -def _repair_anthropic_history(history: list[dict[str, Any]]) -> None: +def _repair_anthropic_history(history: list[Metadata]) -> None: if not history: return last = history[-1] if last.get("role") != "assistant": return @@ -1394,7 +1406,7 @@ def _send_anthropic( md_content: str, user_message: str, base_dir: str, - file_items: list[dict[str, Any]] | None = None, + file_items: list[Metadata] | None = None, discussion_history: str = "", pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None, qa_callback: Optional[Callable[[str], str]] = None, @@ -1424,12 +1436,12 @@ def _send_anthropic( _ensure_anthropic_client() mcp_client.configure(file_items or [], [base_dir]) stable_prompt = _get_combined_system_prompt() - stable_blocks: list[dict[str, Any]] = [{"type": "text", "text": stable_prompt, "cache_control": {"type": "ephemeral"}}] + stable_blocks: list[Metadata] = [{"type": "text", "text": stable_prompt, "cache_control": {"type": "ephemeral"}}] context_text = f"\n\n\n{md_content}\n" context_blocks = _build_chunked_context_blocks(context_text) system_blocks = stable_blocks + context_blocks if discussion_history and not _anthropic_history: - user_content: list[dict[str, Any]] = [{"type": "text", "text": f"[DISCUSSION HISTORY]\n\n{discussion_history}\n\n---\n\n{user_message}"}] + user_content: list[Metadata] = [{"type": "text", "text": f"[DISCUSSION HISTORY]\n\n{discussion_history}\n\n---\n\n{user_message}"}] else: user_content = [{"type": "text", "text": user_message}] for msg in _anthropic_history: @@ -1449,7 +1461,7 @@ def _send_anthropic( all_text_parts: list[str] = [] _cumulative_tool_bytes = 0 - def _strip_private_keys(history: list[dict[str, Any]]) -> list[dict[str, Any]]: + def _strip_private_keys(history: list[Metadata]) -> list[Metadata]: return [{k: v for k, v in m.items() if not k.startswith("_")} for m in history] for round_idx in range(MAX_TOOL_ROUNDS + 2): @@ -1503,7 +1515,7 @@ def _send_anthropic( for b in response.content if getattr(b, "type", None) == "tool_use" ] - usage_dict: dict[str, Any] = {} + usage_dict: Metadata = {} if response.usage: usage_dict["input_tokens"] = response.usage.input_tokens usage_dict["output_tokens"] = response.usage.output_tokens @@ -1532,7 +1544,7 @@ def _send_anthropic( except RuntimeError: results = asyncio.run(_execute_tool_calls_concurrently(response.content, base_dir, pre_tool_callback, qa_callback, round_idx, "anthropic", patch_callback)) - tool_results: list[dict[str, Any]] = [] + tool_results: list[Metadata] = [] for i, (name, call_id, out, _) in enumerate(results): truncated = _truncate_tool_output(out) _cumulative_tool_bytes += len(truncated) @@ -1589,7 +1601,7 @@ def _send_anthropic( #region: Gemini Provider -def get_gemini_cache_stats() -> dict[str, Any]: +def get_gemini_cache_stats() -> Metadata: _ensure_gemini_client() if not _gemini_client: return {"cache_count": 0, "total_size_bytes": 0, "cached_files": []} caches_iterator = _gemini_client.caches.list() @@ -1691,7 +1703,7 @@ def _should_cache_gemini_result(sys_instr: str) -> Result[bool]: errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=f"failed to count gemini tokens: {e}", source="ai_client._should_cache_gemini_result", original=e)], ) -def _create_gemini_cache_result(sys_instr: str, tools_decl: Any, file_items: list[dict[str, Any]] | None) -> Result[Any]: +def _create_gemini_cache_result(sys_instr: str, tools_decl: Any, file_items: list[Metadata] | None) -> Result[Any]: """Create a Gemini cache and the corresponding GenerateContentConfig. Returns Result(data=chat_config_with_cached_content) on success and @@ -1731,7 +1743,7 @@ def _create_gemini_cache_result(sys_instr: str, tools_decl: Any, file_items: lis errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=f"failed to create gemini cache: {type(e).__name__}: {e}", source="ai_client._create_gemini_cache_result", original=e)], ) -def _send_cli_round_result(r_idx: int, adapter: Any, payload: Any, safety_settings: list[Any], sys_instr: str, stream_callback: Optional[Callable[[str], None]]) -> Result[dict[str, Any]]: +def _send_cli_round_result(r_idx: int, adapter: Any, payload: Any, safety_settings: list[Any], sys_instr: str, stream_callback: Optional[Callable[[str], None]]) -> Result[Metadata]: """Call the Gemini CLI adapter for one round. Returns Result[resp_data]. On SDK failure, emits a response_received event with the error info @@ -1788,7 +1800,7 @@ def _get_gemini_history_list(chat: Any | None) -> list[Any]: return [] def _send_gemini(md_content: str, user_message: str, base_dir: str, - file_items: list[dict[str, Any]] | None = None, + file_items: list[Metadata] | None = None, discussion_history: str = "", pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None, qa_callback: Optional[Callable[[str], str]] = None, @@ -1851,7 +1863,7 @@ def _send_gemini(md_content: str, user_message: str, base_dir: str, cached_config_result = _create_gemini_cache_result(sys_instr, tools_decl, file_items) if cached_config_result.ok: chat_config = cached_config_result.data - kwargs: dict[str, Any] = {"model": _model, "config": chat_config} + kwargs: Metadata = {"model": _model, "config": chat_config} if old_history: kwargs["history"] = old_history if _gemini_client: @@ -1957,7 +1969,7 @@ def _send_gemini(md_content: str, user_message: str, base_dir: str, _append_comms("OUT", "request", {"message": f"[GEMINI HISTORY TRIMMED: dropped {dropped} old entries to stay within token budget]"}) if not calls or r_idx > MAX_TOOL_ROUNDS: break f_resps: list[types.Part] = [] - log: list[dict[str, Any]] = [] + log: list[Metadata] = [] # Execute tools concurrently try: @@ -2005,7 +2017,7 @@ def _send_gemini(md_content: str, user_message: str, base_dir: str, return Result(data="", errors=[_classify_gemini_error(e, source="ai_client.gemini")]) def _send_gemini_cli(md_content: str, user_message: str, base_dir: str, - file_items: list[dict[str, Any]] | None = None, + file_items: list[Metadata] | None = None, discussion_history: str = "", pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None, qa_callback: Optional[Callable[[str], str]] = None, @@ -2029,7 +2041,7 @@ def _send_gemini_cli(md_content: str, user_message: str, base_dir: str, mcp_client.configure(file_items or [], [base_dir]) sys_instr = f"{_get_combined_system_prompt()}\n\n\n{md_content}\n" safety_settings = [{'category': 'HARM_CATEGORY_DANGEROUS_CONTENT', 'threshold': 'BLOCK_ONLY_HIGH'}] - payload: Union[str, list[dict[str, Any]]] = user_message + payload: Union[str, list[Metadata]] = user_message if adapter.session_id is None: if discussion_history: payload = f"[DISCUSSION HISTORY]\n\n{discussion_history}\n\n---\n\n{user_message}" @@ -2053,7 +2065,7 @@ def _send_gemini_cli(md_content: str, user_message: str, base_dir: str, usage = adapter.last_usage or {} latency = adapter.last_latency events.emit("response_received", payload={"provider": "gemini_cli", "model": _model, "usage": usage, "latency": latency, "round": r_idx}) - log_calls: list[dict[str, Any]] = [] + log_calls: list[Metadata] = [] for c in calls: log_calls.append({"name": c.get("name"), "args": c.get("args"), "id": c.get("id")}) _append_comms("IN", "response", { @@ -2074,9 +2086,9 @@ def _send_gemini_cli(md_content: str, user_message: str, base_dir: str, }) return NormalizedResponse(text=txt, tool_calls=calls, usage_input_tokens=usage.get("prompt_tokens", 0), usage_output_tokens=usage.get("completion_tokens", 0), usage_cache_read_tokens=0, usage_cache_creation_tokens=0, raw_response=resp_data) - def _pre_dispatch(r_idx: int, calls: list[dict[str, Any]]) -> list[dict[str, Any]]: + def _pre_dispatch(r_idx: int, calls: list[Metadata]) -> list[Metadata]: nonlocal payload, cumulative_tool_bytes, file_items - tool_results_for_cli: list[dict[str, Any]] = [] + tool_results_for_cli: list[Metadata] = [] results_iter: list[tuple[str, str, str, str]] = [] from src.ai_client import _execute_tool_calls_concurrently as _executor try: @@ -2123,7 +2135,7 @@ def _send_gemini_cli(md_content: str, user_message: str, base_dir: str, def _list_deepseek_models(api_key: str) -> list[str]: return ["deepseek-chat", "deepseek-reasoner"] -def _repair_deepseek_history(history: list[dict[str, Any]]) -> None: +def _repair_deepseek_history(history: list[Metadata]) -> None: if not history: return last = history[-1] @@ -2151,7 +2163,7 @@ def _ensure_deepseek_client() -> None: pass def _send_deepseek(md_content: str, user_message: str, base_dir: str, - file_items: list[dict[str, Any]] | None = None, + file_items: list[Metadata] | None = None, discussion_history: str = "", stream: bool = False, pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None, @@ -2198,7 +2210,7 @@ def _send_deepseek(md_content: str, user_message: str, base_dir: str, _cumulative_tool_bytes = 0 for round_idx in range(MAX_TOOL_ROUNDS + 2): - current_api_messages: list[dict[str, Any]] = [] + current_api_messages: list[Metadata] = [] # DeepSeek R1 (Reasoner) can be extremely strict about the 'system' role. # For maximum compatibility, we'll only use 'system' for non-reasoner models. @@ -2235,7 +2247,7 @@ def _send_deepseek(md_content: str, user_message: str, base_dir: str, current_api_messages.append(api_msg) - request_payload: dict[str, Any] = { + request_payload: Metadata = { "model": _model, "messages": current_api_messages, "stream": stream, @@ -2270,9 +2282,9 @@ def _send_deepseek(md_content: str, user_message: str, base_dir: str, if stream: aggregated_content = "" - aggregated_tool_calls: list[dict[str, Any]] = [] + aggregated_tool_calls: list[Metadata] = [] aggregated_reasoning = "" - current_usage: dict[str, Any] = {} + current_usage: Metadata = {} final_finish_reason = "stop" for line in response.iter_lines(): if not line: @@ -2286,9 +2298,9 @@ def _send_deepseek(md_content: str, user_message: str, base_dir: str, chunk = json.loads(chunk_str) if not chunk.get("choices"): if chunk.get("usage"): - current_usage = cast(dict[str, Any], chunk["usage"]) + current_usage = cast(Metadata, chunk["usage"]) continue - delta = cast(dict[str, Any], chunk.get("choices", [{}])[0].get("delta", {})) + delta = cast(Metadata, chunk.get("choices", [{}])[0].get("delta", {})) if delta.get("content"): content_chunk = cast(str, delta["content"]) aggregated_content += content_chunk @@ -2311,7 +2323,7 @@ def _send_deepseek(md_content: str, user_message: str, base_dir: str, if chunk.get("choices", [{}])[0].get("finish_reason"): final_finish_reason = cast(str, chunk["choices"][0]["finish_reason"]) if chunk.get("usage"): - current_usage = cast(dict[str, Any], chunk["usage"]) + current_usage = cast(Metadata, chunk["usage"]) except json.JSONDecodeError: continue assistant_text = aggregated_content @@ -2340,7 +2352,7 @@ def _send_deepseek(md_content: str, user_message: str, base_dir: str, with _deepseek_history_lock: # DeepSeek/OpenAI: If tool_calls are present, content can be null but should usually be present - msg_to_store: dict[str, Any] = {"role": "assistant", "content": assistant_text or None} + msg_to_store: Metadata = {"role": "assistant", "content": assistant_text or None} if reasoning_content: msg_to_store["reasoning_content"] = reasoning_content if tool_calls_raw: @@ -2374,7 +2386,7 @@ def _send_deepseek(md_content: str, user_message: str, base_dir: str, except RuntimeError: results = asyncio.run(_execute_tool_calls_concurrently(tool_calls_raw, base_dir, pre_tool_callback, qa_callback, round_idx, "deepseek", patch_callback)) - tool_results_for_history: list[dict[str, Any]] = [] + tool_results_for_history: list[Metadata] = [] for i, (name, call_id, out, _) in enumerate(results): if i == len(results) - 1: if file_items: @@ -2447,7 +2459,7 @@ def _list_minimax_models_result(api_key: str) -> Result[list[str]]: ) -def _repair_minimax_history(history: list[dict[str, Any]]) -> None: +def _repair_minimax_history(history: list[Metadata]) -> None: if not history: return last = history[-1] if last.get("role") != "assistant": return @@ -2467,7 +2479,7 @@ def _repair_minimax_history(history: list[dict[str, Any]]) -> None: "content": "ERROR: Session was interrupted before tool result was recorded.", }) -def _trim_minimax_history(system_blocks: list[dict[str, Any]], history: list[dict[str, Any]]) -> int: +def _trim_minimax_history(system_blocks: list[Metadata], history: list[Metadata]) -> int: est = _estimate_prompt_tokens(system_blocks, history) limit = 180_000 if est <= limit: @@ -2516,7 +2528,7 @@ def _ensure_grok_client() -> Any: return _grok_client def _send_grok(md_content: str, user_message: str, base_dir: str, - file_items: list[dict[str, Any]] | None = None, + file_items: list[Metadata] | None = None, discussion_history: str = "", stream: bool = False, pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None, @@ -2534,7 +2546,7 @@ def _send_grok(md_content: str, user_message: str, base_dir: str, md_content (str): Markdown formatted context content. user_message (str): User prompt text. base_dir (str): Workspace root directory. - file_items (Optional[list[dict[str, Any]]]): Media or file items for multimodal queries. + file_items (Optional[FileItems]): Media or file items for multimodal queries. discussion_history (str): Contextual discussion text. stream (bool): Whether to stream output. pre_tool_callback (Optional[Callable]): Hook for HITL tool confirmation. @@ -2558,7 +2570,7 @@ def _send_grok(md_content: str, user_message: str, base_dir: str, from src.openai_compatible import OpenAICompatibleRequest, _classify_openai_compatible_error try: client = _ensure_grok_client() - tools: list[dict[str, Any]] | None = _get_deepseek_tools() or None + tools: list[Metadata] | None = _get_deepseek_tools() or None caps = get_capabilities("grok", _model) with _grok_history_lock: user_content = user_message @@ -2572,9 +2584,9 @@ def _send_grok(md_content: str, user_message: str, base_dir: str, _grok_history.append({"role": "user", "content": user_content}) def _build_grok_request(_round_idx: int) -> OpenAICompatibleRequest: with _grok_history_lock: - messages: list[dict[str, Any]] = [{"role": "system", "content": f"{_get_combined_system_prompt()}\n\n\n{md_content}\n"}] + messages: list[Metadata] = [{"role": "system", "content": f"{_get_combined_system_prompt()}\n\n\n{md_content}\n"}] messages.extend(_grok_history) - extra_body: dict[str, Any] = {} + extra_body: Metadata = {} if caps.web_search: extra_body["search_parameters"] = {"mode": "auto"} if caps.x_search: @@ -2600,7 +2612,7 @@ def _list_grok_models() -> list[str]: return list_models_for_vendor("grok") def _send_minimax(md_content: str, user_message: str, base_dir: str, - file_items: list[dict[str, Any]] | None = None, + file_items: list[Metadata] | None = None, discussion_history: str = "", stream: bool = False, pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None, @@ -2618,7 +2630,7 @@ def _send_minimax(md_content: str, user_message: str, base_dir: str, md_content (str): Markdown formatted context content. user_message (str): User prompt text. base_dir (str): Workspace root directory. - file_items (Optional[list[dict[str, Any]]]): Media or file items for multimodal queries. + file_items (Optional[FileItems]): Media or file items for multimodal queries. discussion_history (str): Contextual discussion text. stream (bool): Whether to stream output. pre_tool_callback (Optional[Callable]): Hook for HITL tool confirmation. @@ -2643,7 +2655,7 @@ def _send_minimax(md_content: str, user_message: str, base_dir: str, from src.openai_compatible import OpenAICompatibleRequest try: _ensure_minimax_client() - tools: list[dict[str, Any]] | None = _get_deepseek_tools() or None + tools: list[Metadata] | None = _get_deepseek_tools() or None _repair_minimax_history(_minimax_history) if discussion_history and not _minimax_history: _minimax_history.append({"role": "user", "content": f"[DISCUSSION HISTORY]\n\n{discussion_history}\n\n---\n\n{user_message}"}) @@ -2651,7 +2663,7 @@ def _send_minimax(md_content: str, user_message: str, base_dir: str, _minimax_history.append({"role": "user", "content": user_message}) def _build_minimax_request(_round_idx: int) -> OpenAICompatibleRequest: with _minimax_history_lock: - messages: list[dict[str, Any]] = [{"role": "system", "content": f"{_get_combined_system_prompt()}\n\n\n{md_content}\n"}] + messages: list[Metadata] = [{"role": "system", "content": f"{_get_combined_system_prompt()}\n\n\n{md_content}\n"}] messages.extend(_minimax_history) return OpenAICompatibleRequest( messages=messages, model=_model, temperature=_temperature, top_p=_top_p, @@ -2699,16 +2711,16 @@ def _ensure_qwen_client() -> None: def _dashscope_call( model: str, - messages: list[dict[str, Any]], - tools: list[dict[str, Any]] | None, + messages: list[Metadata], + tools: list[Metadata] | None, *, max_tokens: int, temperature: float, top_p: float, -) -> dict[str, Any]: +) -> Metadata: import dashscope from src.qwen_adapter import build_dashscope_tools - kwargs: dict[str, Any] = { + kwargs: Metadata = { "model": model, "messages": messages, "max_tokens": max_tokens, @@ -2735,8 +2747,8 @@ def _dashscope_exception_from_response(resp: Any) -> Exception: msg = getattr(resp, "message", "unknown dashscope error") return RuntimeError(msg) -def _extract_dashscope_tool_calls(resp: Any) -> list[dict[str, Any]]: - out: list[dict[str, Any]] = [] +def _extract_dashscope_tool_calls(resp: Any) -> list[Metadata]: + out: list[Metadata] = [] if not (hasattr(resp, "output") and resp.output and getattr(resp.output, "tool_calls", None)): return out for tc in resp.output.tool_calls: @@ -2755,7 +2767,7 @@ def _list_qwen_models() -> list[str]: return list_models_for_vendor("qwen") def _send_qwen(md_content: str, user_message: str, base_dir: str, - file_items: list[dict[str, Any]] | None = None, + file_items: list[Metadata] | None = None, discussion_history: str = "", stream: bool = False, pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None, @@ -2773,7 +2785,7 @@ def _send_qwen(md_content: str, user_message: str, base_dir: str, md_content (str): Markdown formatted context content. user_message (str): User prompt text. base_dir (str): Workspace root directory. - file_items (Optional[list[dict[str, Any]]]): Media or file items for multimodal queries. + file_items (Optional[FileItems]): Media or file items for multimodal queries. discussion_history (str): Contextual discussion text. stream (bool): Whether to stream output. pre_tool_callback (Optional[Callable]): Hook for HITL tool confirmation. @@ -2840,7 +2852,7 @@ def _ensure_llama_client() -> Any: return _llama_client def _send_llama(md_content: str, user_message: str, base_dir: str, - file_items: list[dict[str, Any]] | None = None, + file_items: list[Metadata] | None = None, discussion_history: str = "", stream: bool = False, pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None, @@ -2858,7 +2870,7 @@ def _send_llama(md_content: str, user_message: str, base_dir: str, md_content (str): Markdown formatted context content. user_message (str): User prompt text. base_dir (str): Workspace root directory. - file_items (Optional[list[dict[str, Any]]]): Media or file items for multimodal queries. + file_items (Optional[FileItems]): Media or file items for multimodal queries. discussion_history (str): Contextual discussion text. stream (bool): Whether to stream output. pre_tool_callback (Optional[Callable]): Hook for HITL tool confirmation. @@ -2885,7 +2897,7 @@ def _send_llama(md_content: str, user_message: str, base_dir: str, if "localhost" in _llama_base_url or "127.0.0.1" in _llama_base_url: return _send_llama_native(md_content, user_message, base_dir, file_items, discussion_history, stream, pre_tool_callback, qa_callback, stream_callback, patch_callback) client = _ensure_llama_client() - tools: list[dict[str, Any]] | None = _get_deepseek_tools() or None + tools: list[Metadata] | None = _get_deepseek_tools() or None with _llama_history_lock: user_content = user_message if file_items: @@ -2898,7 +2910,7 @@ def _send_llama(md_content: str, user_message: str, base_dir: str, _llama_history.append({"role": "user", "content": user_content}) def _build_llama_request(_round_idx: int) -> OpenAICompatibleRequest: with _llama_history_lock: - messages: list[dict[str, Any]] = [{"role": "system", "content": f"{_get_combined_system_prompt()}\n\n\n{md_content}\n"}] + messages: list[Metadata] = [{"role": "system", "content": f"{_get_combined_system_prompt()}\n\n\n{md_content}\n"}] messages.extend(_llama_history) return OpenAICompatibleRequest( messages=messages, model=_model, temperature=_temperature, top_p=_top_p, @@ -2919,15 +2931,15 @@ OLLAMA_DEFAULT_BASE_URL: str = "http://localhost:11434" def ollama_chat( model: str, - messages: list[dict[str, Any]], + messages: list[Metadata], *, think: str = "low", images: list[str] | None = None, - tools: list[dict[str, Any]] | None = None, + tools: list[Metadata] | None = None, base_url: str = OLLAMA_DEFAULT_BASE_URL, - ) -> dict[str, Any]: + ) -> Metadata: requests = _require_warmed("requests") - payload: dict[str, Any] = {"model": model, "messages": messages, "stream": False} + payload: Metadata = {"model": model, "messages": messages, "stream": False} if think: payload["think"] = think if images: @@ -2938,7 +2950,7 @@ def ollama_chat( return resp.json() def _send_llama_native(md_content: str, user_message: str, base_dir: str, - file_items: list[dict[str, Any]] | None = None, + file_items: list[Metadata] | None = None, discussion_history: str = "", stream: bool = False, pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None, @@ -2956,7 +2968,7 @@ def _send_llama_native(md_content: str, user_message: str, base_dir: str, md_content (str): Markdown formatted context content. user_message (str): User prompt text. base_dir (str): Workspace root directory. - file_items (Optional[list[dict[str, Any]]]): Media or file items for multimodal queries. + file_items (Optional[FileItems]): Media or file items for multimodal queries. discussion_history (str): Contextual discussion text. stream (bool): Whether to stream output. pre_tool_callback (Optional[Callable]): Hook for HITL tool confirmation. @@ -2984,7 +2996,7 @@ def _send_llama_native(md_content: str, user_message: str, base_dir: str, _llama_history.append({"role": "user", "content": f"[DISCUSSION HISTORY]\n\n{discussion_history}\n\n---\n\n{user_message}"}) else: _llama_history.append({"role": "user", "content": user_message}) - messages: list[dict[str, Any]] = [{"role": "system", "content": f"{_get_combined_system_prompt()}\n\n\n{md_content}\n"}] + messages: list[Metadata] = [{"role": "system", "content": f"{_get_combined_system_prompt()}\n\n\n{md_content}\n"}] messages.extend(_llama_history) images: list[str] = [] if file_items: @@ -2995,7 +3007,7 @@ def _send_llama_native(md_content: str, user_message: str, base_dir: str, text = response.get("message", {}).get("content", "") thinking = response.get("message", {}).get("thinking", "") with _llama_history_lock: - msg: dict[str, Any] = {"role": "assistant", "content": text or None} + msg: Metadata = {"role": "assistant", "content": text or None} if thinking: msg["thinking"] = thinking _llama_history.append(msg) @@ -3164,7 +3176,7 @@ def _count_gemini_tokens_for_stats_result(md_content: str) -> Result[int]: ) -def get_token_stats(md_content: str) -> dict[str, Any]: +def get_token_stats(md_content: str) -> Metadata: """ [C: src/app_controller.py:AppController._refresh_api_metrics] """ @@ -3191,7 +3203,7 @@ def send( md_content: str, user_message: str, base_dir: str = ".", - file_items: list[dict[str, Any]] | None = None, + file_items: list[Metadata] | None = None, discussion_history: str = "", stream: bool = False, pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None, @@ -3215,7 +3227,7 @@ def send( md_content (str): System prompt template or markdown prompt structure. user_message (str): The primary user instruction. base_dir (str): Base workspace directory path (defaults to "."). - file_items (list[dict[str, Any]] | None): Optional list of active context files. + file_items (list[Metadata] | None): Optional list of active context files. discussion_history (str): Contextual discussion history lines. stream (bool): Whether to stream the response chunks. pre_tool_callback (Optional[Callable]): Hook called before executing tool calls. @@ -3311,7 +3323,7 @@ def send( if monitor.enabled: monitor.end_component("ai_client.send") return res -def _add_bleed_derived(d: dict[str, Any], sys_tok: int = 0, tool_tok: int = 0) -> dict[str, Any]: +def _add_bleed_derived(d: Metadata, sys_tok: int = 0, tool_tok: int = 0) -> Metadata: """ [C: tests/test_token_viz.py:test_add_bleed_derived_aliases, tests/test_token_viz.py:test_add_bleed_derived_breakdown, tests/test_token_viz.py:test_add_bleed_derived_headroom, tests/test_token_viz.py:test_add_bleed_derived_headroom_clamped_to_zero, tests/test_token_viz.py:test_add_bleed_derived_history_clamped_to_zero, tests/test_token_viz.py:test_add_bleed_derived_would_trim_false, tests/test_token_viz.py:test_add_bleed_derived_would_trim_true, tests/test_token_viz.py:test_would_trim_boundary_exact, tests/test_token_viz.py:test_would_trim_just_above_threshold, tests/test_token_viz.py:test_would_trim_just_below_threshold] """