ed/manual_slop
Private
Public Access
0
0
Fork 0
Code Activity

refactor(ai_client): replace 192 weak type sites with aliases

Browse Source
This commit is contained in:
ed
2026-06-21 12:31:27 -04:00
parent 877bc0f06b
commit 852dea845f
1 changed files with 130 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ai_client.py
+130 -118
View File
@@ -62,6 +62,18 @@ PROVIDERS: List[str] = ["gemini", "anthropic", "gemini_cli", "deepseek", "minima
# hasattr(src.ai_client, '_require_warmed')) continue to work. # hasattr(src.ai_client, '_require_warmed')) continue to work.
from src.module_loader import _require_warmed # noqa: E402,F401 from src.module_loader import _require_warmed # noqa: E402,F401
from src.result_types import ErrorInfo, ErrorKind, Result # 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" _provider: str = "gemini"
_model: str = "gemini-2.5-flash-lite" _model: str = "gemini-2.5-flash-lite"
@@ -96,28 +108,28 @@ _gemini_cached_file_paths: list[str] = []
_GEMINI_CACHE_TTL: int = 3600 _GEMINI_CACHE_TTL: int = 3600
_anthropic_client: Optional[anthropic.Anthropic] = None _anthropic_client: Optional[anthropic.Anthropic] = None
_anthropic_history: list[dict[str, Any]] = [] _anthropic_history: list[Metadata] = []
_anthropic_history_lock: threading.Lock = threading.Lock() _anthropic_history_lock: threading.Lock = threading.Lock()
_deepseek_client: Any = None _deepseek_client: Any = None
_deepseek_history: list[dict[str, Any]] = [] _deepseek_history: list[Metadata] = []
_deepseek_history_lock: threading.Lock = threading.Lock() _deepseek_history_lock: threading.Lock = threading.Lock()
_minimax_client: Any = None _minimax_client: Any = None
_minimax_history: list[dict[str, Any]] = [] _minimax_history: list[Metadata] = []
_minimax_history_lock: threading.Lock = threading.Lock() _minimax_history_lock: threading.Lock = threading.Lock()
_qwen_client: Any = None _qwen_client: Any = None
_qwen_history: list[dict[str, Any]] = [] _qwen_history: list[Metadata] = []
_qwen_history_lock: threading.Lock = threading.Lock() _qwen_history_lock: threading.Lock = threading.Lock()
_qwen_region: str = "china" _qwen_region: str = "china"
_grok_client: Any = None _grok_client: Any = None
_grok_history: list[dict[str, Any]] = [] _grok_history: list[Metadata] = []
_grok_history_lock: threading.Lock = threading.Lock() _grok_history_lock: threading.Lock = threading.Lock()
_llama_client: Any = None _llama_client: Any = None
_llama_history: list[dict[str, Any]] = [] _llama_history: list[Metadata] = []
_llama_history_lock: threading.Lock = threading.Lock() _llama_history_lock: threading.Lock = threading.Lock()
_llama_base_url: str = "http://localhost:11434/v1" _llama_base_url: str = "http://localhost:11434/v1"
_llama_api_key: str = "ollama" _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. # Injected by gui.py - called whenever a comms entry is appended.
# Use get_comms_log_callback/set_comms_log_callback for thread-safe access. # 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. # Injected by gui.py - called whenever a tool call completes.
tool_log_callback: Optional[Callable[[str, str], None]] = None 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: def get_combined_system_prompt(preset: Optional[ToolPreset] = None, bias: Optional[BiasProfile] = None) -> str:
return _get_combined_system_prompt(preset, bias) 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 COMMS_CLAMP_CHARS: int = 300
@@ -232,18 +244,18 @@ COMMS_CLAMP_CHARS: int = 300
#region: Comms Log #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) tl_cb = getattr(_local_storage, "comms_log_callback", None)
if tl_cb: return tl_cb if tl_cb: return tl_cb
return comms_log_callback 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 global comms_log_callback
comms_log_callback = cb comms_log_callback = cb
_local_storage.comms_log_callback = cb _local_storage.comms_log_callback = cb
def _append_comms(direction: str, kind: str, payload: dict[str, Any]) -> None: def _append_comms(direction: str, kind: str, payload: Metadata) -> None:
entry: dict[str, Any] = { entry: Metadata = {
"ts": datetime.datetime.now().strftime("%H:%M:%S"), "ts": datetime.datetime.now().strftime("%H:%M:%S"),
"direction": direction, "direction": direction,
"kind": kind, "kind": kind,
@@ -258,7 +270,7 @@ def _append_comms(direction: str, kind: str, payload: dict[str, Any]) -> None:
if _cb is not None: if _cb is not None:
_cb(entry) _cb(entry)
def get_comms_log() -> list[dict[str, Any]]: def get_comms_log() -> list[Metadata]:
return list(_comms_log) return list(_comms_log)
def clear_comms_log() -> None: def clear_comms_log() -> None:
@@ -267,7 +279,7 @@ def clear_comms_log() -> None:
def get_credentials_path() -> Path: def get_credentials_path() -> Path:
return Path(os.environ.get("SLOP_CREDENTIALS", str(Path(__file__).parent.parent / "credentials.toml"))) 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() cred_path = get_credentials_path()
try: try:
with open(cred_path, "rb") as f: 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.""" """Returns the name of the currently active bias profile."""
return _active_bias_profile.name if _active_bias_profile else None 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] [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(): for spec in mcp_client.get_tool_schemas():
if _agent_tools.get(spec["name"], True): if _agent_tools.get(spec["name"], True):
raw_tools.append({ raw_tools.append({
@@ -647,9 +659,9 @@ def _build_anthropic_tools() -> list[dict[str, Any]]:
raw_tools[-1]["cache_control"] = {"type": "ephemeral"} raw_tools[-1]["cache_control"] = {"type": "ephemeral"}
return raw_tools 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] [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. # completes the chain once, then `.types` is just an attribute access.
genai = _require_warmed("google.genai") genai = _require_warmed("google.genai")
types = genai.types types = genai.types
raw_tools: list[dict[str, Any]] = [] raw_tools: list[Metadata] = []
for spec in mcp_client.get_tool_schemas(): for spec in mcp_client.get_tool_schemas():
if _agent_tools.get(spec["name"], True): if _agent_tools.get(spec["name"], True):
raw_tools.append({ raw_tools.append({
@@ -726,7 +738,7 @@ def _gemini_tool_declaration() -> Optional[types.Tool]:
#region: Tool Execution #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]. """Parse tool call arguments from JSON. Returns Result[dict, ErrorInfo].
On JSON parse failure, returns Result(data={}, errors=[ErrorInfo(...)]). On JSON parse failure, returns Result(data={}, errors=[ErrorInfo(...)]).
@@ -789,8 +801,8 @@ async def _execute_tool_calls_concurrently(
tasks = [] tasks = []
for fc in calls: 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 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 == "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(dict[str, Any], getattr(fc, "input")), cast(str, getattr(fc, "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": elif provider == "deepseek":
tool_info = fc.get("function", {}) tool_info = fc.get("function", {})
name = cast(str, tool_info.get("name")) name = cast(str, tool_info.get("name"))
@@ -830,11 +842,11 @@ def run_with_tool_loop(
base_dir: str, base_dir: str,
vendor_name: str, vendor_name: str,
history_lock: Optional[threading.Lock] = None, history_lock: Optional[threading.Lock] = None,
history: Optional[list[dict[str, Any]]] = None, history: Optional[FileItems] = None,
trim_func: Optional[Callable[[list[dict[str, Any]]], None]] = None, trim_func: Optional[Callable[[list[Metadata]], None]] = None,
reasoning_extractor: Optional[Callable[[Any], str]] = None, reasoning_extractor: Optional[Callable[[Any], str]] = None,
send_func: Optional[Callable[[int], NormalizedResponse]] = 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, wrap_reasoning_in_text: bool = False,
) -> str: ) -> str:
""" """
@@ -855,7 +867,7 @@ def run_with_tool_loop(
base_dir (str): Base workspace directory. base_dir (str): Base workspace directory.
vendor_name (str): The vendor name. vendor_name (str): The vendor name.
history_lock (Optional[threading.Lock]): Lock for thread safety on history. 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. trim_func (Optional[Callable]): Trimming callback for history.
reasoning_extractor (Optional[Callable]): Callback to extract reasoning content. reasoning_extractor (Optional[Callable]): Callback to extract reasoning content.
send_func (Optional[Callable]): Dispatch sender callback. send_func (Optional[Callable]): Dispatch sender callback.
@@ -898,7 +910,7 @@ def run_with_tool_loop(
response_text = response.text or "" response_text = response.text or ""
if history_lock is not None and history is not None: if history_lock is not None and history is not None:
with history_lock: 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 reasoning_content: msg["reasoning_content"] = reasoning_content
if response.tool_calls: msg["tool_calls"] = response.tool_calls if response.tool_calls: msg["tool_calls"] = response.tool_calls
history.append(msg) history.append(msg)
@@ -932,7 +944,7 @@ def run_with_tool_loop(
async def _execute_single_tool_call_async( async def _execute_single_tool_call_async(
name: str, name: str,
args: dict[str, Any], args: Metadata,
call_id: str, call_id: str,
base_dir: str, base_dir: str,
pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[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: Parameters & Inputs:
name (str): The name of the tool to execute. 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. call_id (str): Unique call identifier.
base_dir (str): Workspace root directory. base_dir (str): Workspace root directory.
pre_tool_callback (Optional[Callable]): Hook for HITL validation. pre_tool_callback (Optional[Callable]): Hook for HITL validation.
@@ -1041,7 +1053,7 @@ def _truncate_tool_output(output: str) -> str:
#region: File Context Building #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. """Re-reads file items, returns (refreshed, changed) tuple.
Per-file read errors are accumulated into Result.errors (structured 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 future callers should check result.errors to detect file re-read
failures. failures.
""" """
refreshed: list[dict[str, Any]] = [] refreshed: list[Metadata] = []
changed: list[dict[str, Any]] = [] changed: list[Metadata] = []
errors: list[ErrorInfo] = [] errors: list[ErrorInfo] = []
for item in file_items: for item in file_items:
path = item.get("path") 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) 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: if not file_items:
return "" return ""
parts: list[str] = [] parts: list[str] = []
@@ -1090,7 +1102,7 @@ def _build_file_context_text(file_items: list[dict[str, Any]]) -> str:
_DIFF_LINE_THRESHOLD: int = 200 _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. 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. the full file is dumped; otherwise, a unified diff is constructed.
Parameters & Inputs: 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: Returns:
str: Combined markdown string representing the changes or full files. 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)") else: parts.append(f"### `{path}` (no changes detected)")
return "\n\n---\n\n".join(parts) return "\n\n---\n\n".join(parts)
def _build_deepseek_tools() -> list[dict[str, Any]]: def _build_deepseek_tools() -> list[ToolDefinition]:
raw_tools: list[dict[str, Any]] = [] raw_tools: list[Metadata] = []
for spec in mcp_client.get_tool_schemas(): for spec in mcp_client.get_tool_schemas():
if _agent_tools.get(spec["name"], True): if _agent_tools.get(spec["name"], True):
raw_tools.append({ raw_tools.append({
@@ -1165,7 +1177,7 @@ def _build_deepseek_tools() -> list[dict[str, Any]]:
}) })
if _active_tool_preset: if _active_tool_preset:
_BIAS_ENGINE.apply_semantic_nudges(raw_tools, _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: for tool_def in raw_tools:
tools_list.append({ tools_list.append({
"type": "function", "type": "function",
@@ -1177,18 +1189,18 @@ def _build_deepseek_tools() -> list[dict[str, Any]]:
}) })
return tools_list 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 global _CACHED_DEEPSEEK_TOOLS
if _CACHED_DEEPSEEK_TOOLS is None: if _CACHED_DEEPSEEK_TOOLS is None:
_CACHED_DEEPSEEK_TOOLS = _build_deepseek_tools() _CACHED_DEEPSEEK_TOOLS = _build_deepseek_tools()
return _CACHED_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 isinstance(block, dict): return block
if hasattr(block, "model_dump"): return cast(dict[str, Any], block.model_dump()) if hasattr(block, "model_dump"): return cast(Metadata, block.model_dump())
if hasattr(block, "to_dict"): return cast(dict[str, Any], block.to_dict()) if hasattr(block, "to_dict"): return cast(Metadata, block.to_dict())
block_type = getattr(block, "type", None) block_type = getattr(block, "type", None)
if block_type == "text": return {"type": "text", "text": block.text} 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")} 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 _GEMINI_MAX_INPUT_TOKENS: int = 900_000
_FILE_REFRESH_MARKER: str = _project_context_marker if _project_context_marker.strip() else "[SYSTEM: FILES UPDATED]" _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") cached = msg.get("_est_tokens")
if cached is not None: return cast(int, cached) if cached is not None: return cast(int, cached)
total_chars = 0 total_chars = 0
@@ -1225,10 +1237,10 @@ def _estimate_message_tokens(msg: dict[str, Any]) -> int:
msg["_est_tokens"] = est msg["_est_tokens"] = est
return est return est
def _invalidate_token_estimate(msg: dict[str, Any]) -> None: def _invalidate_token_estimate(msg: Metadata) -> None:
msg.pop("_est_tokens", 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 total = 0
for block in system_blocks: for block in system_blocks:
text = cast(str, block.get("text", "")) 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) total += _estimate_message_tokens(msg)
return total 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: if len(history) < 2:
return return
last_user_idx = -1 last_user_idx = -1
@@ -1252,7 +1264,7 @@ def _strip_stale_file_refreshes(history: list[dict[str, Any]]) -> None:
content = msg.get("content") content = msg.get("content")
if not isinstance(content, list): if not isinstance(content, list):
continue continue
cleaned: list[dict[str, Any]] = [] cleaned: list[Metadata] = []
for block in content: for block in content:
if isinstance(block, dict) and block.get("type") == "text": if isinstance(block, dict) and block.get("type") == "text":
text = cast(str, block.get("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]: 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)] 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) chunks = _chunk_text(md_content, _ANTHROPIC_CHUNK_SIZE)
blocks: list[dict[str, Any]] = [] blocks: list[Metadata] = []
for i, chunk in enumerate(chunks): 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: if i == len(chunks) - 1:
block["cache_control"] = {"type": "ephemeral"} block["cache_control"] = {"type": "ephemeral"}
blocks.append(block) blocks.append(block)
return blocks return blocks
def _strip_cache_controls(history: list[dict[str, Any]]) -> None: def _strip_cache_controls(history: list[Metadata]) -> None:
for msg in history: for msg in history:
content = msg.get("content") content = msg.get("content")
if isinstance(content, list): if isinstance(content, list):
@@ -1284,7 +1296,7 @@ def _strip_cache_controls(history: list[dict[str, Any]]) -> None:
if isinstance(block, dict): if isinstance(block, dict):
block.pop("cache_control", None) 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"] user_indices = [i for i, m in enumerate(history) if m.get("role") == "user"]
if len(user_indices) < 2: return if len(user_indices) < 2: return
target_idx = user_indices[-2] target_idx = user_indices[-2]
@@ -1338,7 +1350,7 @@ def _ensure_anthropic_client() -> None:
default_headers = {"anthropic-beta": "prompt-caching-2024-07-31"} 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) _strip_stale_file_refreshes(history)
est = _estimate_prompt_tokens(system_blocks, history) est = _estimate_prompt_tokens(system_blocks, history)
if est <= _ANTHROPIC_MAX_PROMPT_TOKENS: return 0 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) est -= _estimate_message_tokens(removed)
return dropped return dropped
def _repair_anthropic_history(history: list[dict[str, Any]]) -> None: def _repair_anthropic_history(history: list[Metadata]) -> None:
if not history: return if not history: return
last = history[-1] last = history[-1]
if last.get("role") != "assistant": return if last.get("role") != "assistant": return
@@ -1394,7 +1406,7 @@ def _send_anthropic(
md_content: str, md_content: str,
user_message: str, user_message: str,
base_dir: str, base_dir: str,
file_items: list[dict[str, Any]] | None = None, file_items: list[Metadata] | None = None,
discussion_history: str = "", discussion_history: str = "",
pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None, pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None,
qa_callback: Optional[Callable[[str], str]] = None, qa_callback: Optional[Callable[[str], str]] = None,
@@ -1424,12 +1436,12 @@ def _send_anthropic(
_ensure_anthropic_client() _ensure_anthropic_client()
mcp_client.configure(file_items or [], [base_dir]) mcp_client.configure(file_items or [], [base_dir])
stable_prompt = _get_combined_system_prompt() 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<context>\n{md_content}\n</context>" context_text = f"\n\n<context>\n{md_content}\n</context>"
context_blocks = _build_chunked_context_blocks(context_text) context_blocks = _build_chunked_context_blocks(context_text)
system_blocks = stable_blocks + context_blocks system_blocks = stable_blocks + context_blocks
if discussion_history and not _anthropic_history: 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: else:
user_content = [{"type": "text", "text": user_message}] user_content = [{"type": "text", "text": user_message}]
for msg in _anthropic_history: for msg in _anthropic_history:
@@ -1449,7 +1461,7 @@ def _send_anthropic(
all_text_parts: list[str] = [] all_text_parts: list[str] = []
_cumulative_tool_bytes = 0 _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] 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): for round_idx in range(MAX_TOOL_ROUNDS + 2):
@@ -1503,7 +1515,7 @@ def _send_anthropic(
for b in response.content for b in response.content
if getattr(b, "type", None) == "tool_use" if getattr(b, "type", None) == "tool_use"
] ]
usage_dict: dict[str, Any] = {} usage_dict: Metadata = {}
if response.usage: if response.usage:
usage_dict["input_tokens"] = response.usage.input_tokens usage_dict["input_tokens"] = response.usage.input_tokens
usage_dict["output_tokens"] = response.usage.output_tokens usage_dict["output_tokens"] = response.usage.output_tokens
@@ -1532,7 +1544,7 @@ def _send_anthropic(
except RuntimeError: except RuntimeError:
results = asyncio.run(_execute_tool_calls_concurrently(response.content, base_dir, pre_tool_callback, qa_callback, round_idx, "anthropic", patch_callback)) 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): for i, (name, call_id, out, _) in enumerate(results):
truncated = _truncate_tool_output(out) truncated = _truncate_tool_output(out)
_cumulative_tool_bytes += len(truncated) _cumulative_tool_bytes += len(truncated)
@@ -1589,7 +1601,7 @@ def _send_anthropic(
#region: Gemini Provider #region: Gemini Provider
def get_gemini_cache_stats() -> dict[str, Any]: def get_gemini_cache_stats() -> Metadata:
_ensure_gemini_client() _ensure_gemini_client()
if not _gemini_client: return {"cache_count": 0, "total_size_bytes": 0, "cached_files": []} if not _gemini_client: return {"cache_count": 0, "total_size_bytes": 0, "cached_files": []}
caches_iterator = _gemini_client.caches.list() 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)], 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. """Create a Gemini cache and the corresponding GenerateContentConfig.
Returns Result(data=chat_config_with_cached_content) on success and 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)], 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]. """Call the Gemini CLI adapter for one round. Returns Result[resp_data].
On SDK failure, emits a response_received event with the error info 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 [] return []
def _send_gemini(md_content: str, user_message: str, base_dir: str, 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 = "", discussion_history: str = "",
pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None, pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None,
qa_callback: Optional[Callable[[str], 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) cached_config_result = _create_gemini_cache_result(sys_instr, tools_decl, file_items)
if cached_config_result.ok: if cached_config_result.ok:
chat_config = cached_config_result.data 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: if old_history:
kwargs["history"] = old_history kwargs["history"] = old_history
if _gemini_client: 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]"}) _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 if not calls or r_idx > MAX_TOOL_ROUNDS: break
f_resps: list[types.Part] = [] f_resps: list[types.Part] = []
log: list[dict[str, Any]] = [] log: list[Metadata] = []
# Execute tools concurrently # Execute tools concurrently
try: 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")]) 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, 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 = "", discussion_history: str = "",
pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None, pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None,
qa_callback: Optional[Callable[[str], 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]) mcp_client.configure(file_items or [], [base_dir])
sys_instr = f"{_get_combined_system_prompt()}\n\n<context>\n{md_content}\n</context>" sys_instr = f"{_get_combined_system_prompt()}\n\n<context>\n{md_content}\n</context>"
safety_settings = [{'category': 'HARM_CATEGORY_DANGEROUS_CONTENT', 'threshold': 'BLOCK_ONLY_HIGH'}] 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 adapter.session_id is None:
if discussion_history: if discussion_history:
payload = f"[DISCUSSION HISTORY]\n\n{discussion_history}\n\n---\n\n{user_message}" 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 {} usage = adapter.last_usage or {}
latency = adapter.last_latency latency = adapter.last_latency
events.emit("response_received", payload={"provider": "gemini_cli", "model": _model, "usage": usage, "latency": latency, "round": r_idx}) 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: for c in calls:
log_calls.append({"name": c.get("name"), "args": c.get("args"), "id": c.get("id")}) log_calls.append({"name": c.get("name"), "args": c.get("args"), "id": c.get("id")})
_append_comms("IN", "response", { _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) 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 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]] = [] results_iter: list[tuple[str, str, str, str]] = []
from src.ai_client import _execute_tool_calls_concurrently as _executor from src.ai_client import _execute_tool_calls_concurrently as _executor
try: 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]: def _list_deepseek_models(api_key: str) -> list[str]:
return ["deepseek-chat", "deepseek-reasoner"] 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: if not history:
return return
last = history[-1] last = history[-1]
@@ -2151,7 +2163,7 @@ def _ensure_deepseek_client() -> None:
pass pass
def _send_deepseek(md_content: str, user_message: str, base_dir: str, 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 = "", discussion_history: str = "",
stream: bool = False, stream: bool = False,
pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None, 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 _cumulative_tool_bytes = 0
for round_idx in range(MAX_TOOL_ROUNDS + 2): 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. # DeepSeek R1 (Reasoner) can be extremely strict about the 'system' role.
# For maximum compatibility, we'll only use 'system' for non-reasoner models. # 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) current_api_messages.append(api_msg)
request_payload: dict[str, Any] = { request_payload: Metadata = {
"model": _model, "model": _model,
"messages": current_api_messages, "messages": current_api_messages,
"stream": stream, "stream": stream,
@@ -2270,9 +2282,9 @@ def _send_deepseek(md_content: str, user_message: str, base_dir: str,
if stream: if stream:
aggregated_content = "" aggregated_content = ""
aggregated_tool_calls: list[dict[str, Any]] = [] aggregated_tool_calls: list[Metadata] = []
aggregated_reasoning = "" aggregated_reasoning = ""
current_usage: dict[str, Any] = {} current_usage: Metadata = {}
final_finish_reason = "stop" final_finish_reason = "stop"
for line in response.iter_lines(): for line in response.iter_lines():
if not line: if not line:
@@ -2286,9 +2298,9 @@ def _send_deepseek(md_content: str, user_message: str, base_dir: str,
chunk = json.loads(chunk_str) chunk = json.loads(chunk_str)
if not chunk.get("choices"): if not chunk.get("choices"):
if chunk.get("usage"): if chunk.get("usage"):
current_usage = cast(dict[str, Any], chunk["usage"]) current_usage = cast(Metadata, chunk["usage"])
continue 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"): if delta.get("content"):
content_chunk = cast(str, delta["content"]) content_chunk = cast(str, delta["content"])
aggregated_content += content_chunk 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"): if chunk.get("choices", [{}])[0].get("finish_reason"):
final_finish_reason = cast(str, chunk["choices"][0]["finish_reason"]) final_finish_reason = cast(str, chunk["choices"][0]["finish_reason"])
if chunk.get("usage"): if chunk.get("usage"):
current_usage = cast(dict[str, Any], chunk["usage"]) current_usage = cast(Metadata, chunk["usage"])
except json.JSONDecodeError: except json.JSONDecodeError:
continue continue
assistant_text = aggregated_content assistant_text = aggregated_content
@@ -2340,7 +2352,7 @@ def _send_deepseek(md_content: str, user_message: str, base_dir: str,
with _deepseek_history_lock: with _deepseek_history_lock:
# DeepSeek/OpenAI: If tool_calls are present, content can be null but should usually be present # 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: if reasoning_content:
msg_to_store["reasoning_content"] = reasoning_content msg_to_store["reasoning_content"] = reasoning_content
if tool_calls_raw: if tool_calls_raw:
@@ -2374,7 +2386,7 @@ def _send_deepseek(md_content: str, user_message: str, base_dir: str,
except RuntimeError: except RuntimeError:
results = asyncio.run(_execute_tool_calls_concurrently(tool_calls_raw, base_dir, pre_tool_callback, qa_callback, round_idx, "deepseek", patch_callback)) 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): for i, (name, call_id, out, _) in enumerate(results):
if i == len(results) - 1: if i == len(results) - 1:
if file_items: 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 if not history: return
last = history[-1] last = history[-1]
if last.get("role") != "assistant": return 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.", "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) est = _estimate_prompt_tokens(system_blocks, history)
limit = 180_000 limit = 180_000
if est <= limit: if est <= limit:
@@ -2516,7 +2528,7 @@ def _ensure_grok_client() -> Any:
return _grok_client return _grok_client
def _send_grok(md_content: str, user_message: str, base_dir: str, 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 = "", discussion_history: str = "",
stream: bool = False, stream: bool = False,
pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None, 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. md_content (str): Markdown formatted context content.
user_message (str): User prompt text. user_message (str): User prompt text.
base_dir (str): Workspace root directory. 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. discussion_history (str): Contextual discussion text.
stream (bool): Whether to stream output. stream (bool): Whether to stream output.
pre_tool_callback (Optional[Callable]): Hook for HITL tool confirmation. 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 from src.openai_compatible import OpenAICompatibleRequest, _classify_openai_compatible_error
try: try:
client = _ensure_grok_client() 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) caps = get_capabilities("grok", _model)
with _grok_history_lock: with _grok_history_lock:
user_content = user_message 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}) _grok_history.append({"role": "user", "content": user_content})
def _build_grok_request(_round_idx: int) -> OpenAICompatibleRequest: def _build_grok_request(_round_idx: int) -> OpenAICompatibleRequest:
with _grok_history_lock: with _grok_history_lock:
messages: list[dict[str, Any]] = [{"role": "system", "content": f"{_get_combined_system_prompt()}\n\n<context>\n{md_content}\n</context>"}] messages: list[Metadata] = [{"role": "system", "content": f"{_get_combined_system_prompt()}\n\n<context>\n{md_content}\n</context>"}]
messages.extend(_grok_history) messages.extend(_grok_history)
extra_body: dict[str, Any] = {} extra_body: Metadata = {}
if caps.web_search: if caps.web_search:
extra_body["search_parameters"] = {"mode": "auto"} extra_body["search_parameters"] = {"mode": "auto"}
if caps.x_search: if caps.x_search:
@@ -2600,7 +2612,7 @@ def _list_grok_models() -> list[str]:
return list_models_for_vendor("grok") return list_models_for_vendor("grok")
def _send_minimax(md_content: str, user_message: str, base_dir: str, 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 = "", discussion_history: str = "",
stream: bool = False, stream: bool = False,
pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None, 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. md_content (str): Markdown formatted context content.
user_message (str): User prompt text. user_message (str): User prompt text.
base_dir (str): Workspace root directory. 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. discussion_history (str): Contextual discussion text.
stream (bool): Whether to stream output. stream (bool): Whether to stream output.
pre_tool_callback (Optional[Callable]): Hook for HITL tool confirmation. 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 from src.openai_compatible import OpenAICompatibleRequest
try: try:
_ensure_minimax_client() _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) _repair_minimax_history(_minimax_history)
if discussion_history and not _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}"}) _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}) _minimax_history.append({"role": "user", "content": user_message})
def _build_minimax_request(_round_idx: int) -> OpenAICompatibleRequest: def _build_minimax_request(_round_idx: int) -> OpenAICompatibleRequest:
with _minimax_history_lock: with _minimax_history_lock:
messages: list[dict[str, Any]] = [{"role": "system", "content": f"{_get_combined_system_prompt()}\n\n<context>\n{md_content}\n</context>"}] messages: list[Metadata] = [{"role": "system", "content": f"{_get_combined_system_prompt()}\n\n<context>\n{md_content}\n</context>"}]
messages.extend(_minimax_history) messages.extend(_minimax_history)
return OpenAICompatibleRequest( return OpenAICompatibleRequest(
messages=messages, model=_model, temperature=_temperature, top_p=_top_p, messages=messages, model=_model, temperature=_temperature, top_p=_top_p,
@@ -2699,16 +2711,16 @@ def _ensure_qwen_client() -> None:
def _dashscope_call( def _dashscope_call(
model: str, model: str,
messages: list[dict[str, Any]], messages: list[Metadata],
tools: list[dict[str, Any]] | None, tools: list[Metadata] | None,
*, *,
max_tokens: int, max_tokens: int,
temperature: float, temperature: float,
top_p: float, top_p: float,
) -> dict[str, Any]: ) -> Metadata:
import dashscope import dashscope
from src.qwen_adapter import build_dashscope_tools from src.qwen_adapter import build_dashscope_tools
kwargs: dict[str, Any] = { kwargs: Metadata = {
"model": model, "model": model,
"messages": messages, "messages": messages,
"max_tokens": max_tokens, "max_tokens": max_tokens,
@@ -2735,8 +2747,8 @@ def _dashscope_exception_from_response(resp: Any) -> Exception:
msg = getattr(resp, "message", "unknown dashscope error") msg = getattr(resp, "message", "unknown dashscope error")
return RuntimeError(msg) return RuntimeError(msg)
def _extract_dashscope_tool_calls(resp: Any) -> list[dict[str, Any]]: def _extract_dashscope_tool_calls(resp: Any) -> list[Metadata]:
out: list[dict[str, Any]] = [] out: list[Metadata] = []
if not (hasattr(resp, "output") and resp.output and getattr(resp.output, "tool_calls", None)): if not (hasattr(resp, "output") and resp.output and getattr(resp.output, "tool_calls", None)):
return out return out
for tc in resp.output.tool_calls: for tc in resp.output.tool_calls:
@@ -2755,7 +2767,7 @@ def _list_qwen_models() -> list[str]:
return list_models_for_vendor("qwen") return list_models_for_vendor("qwen")
def _send_qwen(md_content: str, user_message: str, base_dir: str, 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 = "", discussion_history: str = "",
stream: bool = False, stream: bool = False,
pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None, 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. md_content (str): Markdown formatted context content.
user_message (str): User prompt text. user_message (str): User prompt text.
base_dir (str): Workspace root directory. 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. discussion_history (str): Contextual discussion text.
stream (bool): Whether to stream output. stream (bool): Whether to stream output.
pre_tool_callback (Optional[Callable]): Hook for HITL tool confirmation. pre_tool_callback (Optional[Callable]): Hook for HITL tool confirmation.
@@ -2840,7 +2852,7 @@ def _ensure_llama_client() -> Any:
return _llama_client return _llama_client
def _send_llama(md_content: str, user_message: str, base_dir: str, 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 = "", discussion_history: str = "",
stream: bool = False, stream: bool = False,
pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None, 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. md_content (str): Markdown formatted context content.
user_message (str): User prompt text. user_message (str): User prompt text.
base_dir (str): Workspace root directory. 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. discussion_history (str): Contextual discussion text.
stream (bool): Whether to stream output. stream (bool): Whether to stream output.
pre_tool_callback (Optional[Callable]): Hook for HITL tool confirmation. 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: 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) 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() 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: with _llama_history_lock:
user_content = user_message user_content = user_message
if file_items: 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}) _llama_history.append({"role": "user", "content": user_content})
def _build_llama_request(_round_idx: int) -> OpenAICompatibleRequest: def _build_llama_request(_round_idx: int) -> OpenAICompatibleRequest:
with _llama_history_lock: with _llama_history_lock:
messages: list[dict[str, Any]] = [{"role": "system", "content": f"{_get_combined_system_prompt()}\n\n<context>\n{md_content}\n</context>"}] messages: list[Metadata] = [{"role": "system", "content": f"{_get_combined_system_prompt()}\n\n<context>\n{md_content}\n</context>"}]
messages.extend(_llama_history) messages.extend(_llama_history)
return OpenAICompatibleRequest( return OpenAICompatibleRequest(
messages=messages, model=_model, temperature=_temperature, top_p=_top_p, 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( def ollama_chat(
model: str, model: str,
messages: list[dict[str, Any]], messages: list[Metadata],
*, *,
think: str = "low", think: str = "low",
images: list[str] | None = None, images: list[str] | None = None,
tools: list[dict[str, Any]] | None = None, tools: list[Metadata] | None = None,
base_url: str = OLLAMA_DEFAULT_BASE_URL, base_url: str = OLLAMA_DEFAULT_BASE_URL,
) -> dict[str, Any]: ) -> Metadata:
requests = _require_warmed("requests") requests = _require_warmed("requests")
payload: dict[str, Any] = {"model": model, "messages": messages, "stream": False} payload: Metadata = {"model": model, "messages": messages, "stream": False}
if think: if think:
payload["think"] = think payload["think"] = think
if images: if images:
@@ -2938,7 +2950,7 @@ def ollama_chat(
return resp.json() return resp.json()
def _send_llama_native(md_content: str, user_message: str, base_dir: str, 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 = "", discussion_history: str = "",
stream: bool = False, stream: bool = False,
pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None, 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. md_content (str): Markdown formatted context content.
user_message (str): User prompt text. user_message (str): User prompt text.
base_dir (str): Workspace root directory. 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. discussion_history (str): Contextual discussion text.
stream (bool): Whether to stream output. stream (bool): Whether to stream output.
pre_tool_callback (Optional[Callable]): Hook for HITL tool confirmation. 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}"}) _llama_history.append({"role": "user", "content": f"[DISCUSSION HISTORY]\n\n{discussion_history}\n\n---\n\n{user_message}"})
else: else:
_llama_history.append({"role": "user", "content": user_message}) _llama_history.append({"role": "user", "content": user_message})
messages: list[dict[str, Any]] = [{"role": "system", "content": f"{_get_combined_system_prompt()}\n\n<context>\n{md_content}\n</context>"}] messages: list[Metadata] = [{"role": "system", "content": f"{_get_combined_system_prompt()}\n\n<context>\n{md_content}\n</context>"}]
messages.extend(_llama_history) messages.extend(_llama_history)
images: list[str] = [] images: list[str] = []
if file_items: 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", "") text = response.get("message", {}).get("content", "")
thinking = response.get("message", {}).get("thinking", "") thinking = response.get("message", {}).get("thinking", "")
with _llama_history_lock: with _llama_history_lock:
msg: dict[str, Any] = {"role": "assistant", "content": text or None} msg: Metadata = {"role": "assistant", "content": text or None}
if thinking: if thinking:
msg["thinking"] = thinking msg["thinking"] = thinking
_llama_history.append(msg) _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] [C: src/app_controller.py:AppController._refresh_api_metrics]
""" """
@@ -3191,7 +3203,7 @@ def send(
md_content: str, md_content: str,
user_message: str, user_message: str,
base_dir: str = ".", base_dir: str = ".",
file_items: list[dict[str, Any]] | None = None, file_items: list[Metadata] | None = None,
discussion_history: str = "", discussion_history: str = "",
stream: bool = False, stream: bool = False,
pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None, 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. md_content (str): System prompt template or markdown prompt structure.
user_message (str): The primary user instruction. user_message (str): The primary user instruction.
base_dir (str): Base workspace directory path (defaults to "."). 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. discussion_history (str): Contextual discussion history lines.
stream (bool): Whether to stream the response chunks. stream (bool): Whether to stream the response chunks.
pre_tool_callback (Optional[Callable]): Hook called before executing tool calls. 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") if monitor.enabled: monitor.end_component("ai_client.send")
return res 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] [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]
""" """