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

ai client pass (in gemini region)

Browse Source
This commit is contained in:
ed
2026-06-13 20:49:37 -04:00
parent 94ab6dcc6f
commit 5030bd848f
1 changed files with 228 additions and 275 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ai_client.py
+228 -275
View File
@@ -500,29 +500,29 @@ def set_tool_preset(preset_name: Optional[str]) -> None:
_tool_approval_modes = {} _tool_approval_modes = {}
if not preset_name or preset_name == "None": if not preset_name or preset_name == "None":
# Enable all tools if no preset # Enable all tools if no preset
_agent_tools = {name: True for name in mcp_client.TOOL_NAMES} _agent_tools = {name: True for name in mcp_client.TOOL_NAMES}
_agent_tools[TOOL_NAME] = True _agent_tools[TOOL_NAME] = True
_active_tool_preset = None _active_tool_preset = None
else: else:
try: try:
manager = ToolPresetManager() manager = ToolPresetManager()
presets = manager.load_all() presets = manager.load_all()
if preset_name in presets: if preset_name in presets:
preset = presets[preset_name] preset = presets[preset_name]
_active_tool_preset = preset _active_tool_preset = preset
new_tools = {name: False for name in mcp_client.TOOL_NAMES} new_tools = {name: False for name in mcp_client.TOOL_NAMES}
new_tools[TOOL_NAME] = False new_tools[TOOL_NAME] = False
for cat in preset.categories.values(): for cat in preset.categories.values():
for tool in cat: for tool in cat:
name = tool.name name = tool.name
new_tools[name] = True new_tools[name] = True
_tool_approval_modes[name] = tool.approval _tool_approval_modes[name] = tool.approval
_agent_tools = new_tools _agent_tools = new_tools
except Exception as e: except Exception as e:
sys.stderr.write(f"[ERROR] Failed to set tool preset '{preset_name}': {e}\n") sys.stderr.write(f"[ERROR] Failed to set tool preset '{preset_name}': {e}\n")
sys.stderr.flush() sys.stderr.flush()
_CACHED_ANTHROPIC_TOOLS = None _CACHED_ANTHROPIC_TOOLS = None
_CACHED_DEEPSEEK_TOOLS = None _CACHED_DEEPSEEK_TOOLS = None
def set_bias_profile(profile_name: Optional[str]) -> None: def set_bias_profile(profile_name: Optional[str]) -> None:
"""Sets the active tool bias profile for tuning model behavior.""" """Sets the active tool bias profile for tuning model behavior."""
@@ -531,7 +531,7 @@ def set_bias_profile(profile_name: Optional[str]) -> None:
_active_bias_profile = None _active_bias_profile = None
else: else:
try: try:
manager = ToolPresetManager() manager = ToolPresetManager()
profiles = manager.load_all_bias_profiles() profiles = manager.load_all_bias_profiles()
if profile_name in profiles: if profile_name in profiles:
_active_bias_profile = profiles[profile_name] _active_bias_profile = profiles[profile_name]
@@ -551,8 +551,8 @@ def _build_anthropic_tools() -> list[dict[str, Any]]:
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({
"name": spec["name"], "name": spec["name"],
"description": spec["description"], "description": spec["description"],
"input_schema": spec["parameters"], "input_schema": spec["parameters"],
}) })
if _agent_tools.get(TOOL_NAME, True): if _agent_tools.get(TOOL_NAME, True):
@@ -569,7 +569,7 @@ def _build_anthropic_tools() -> list[dict[str, Any]]:
"type": "object", "type": "object",
"properties": { "properties": {
"script": { "script": {
"type": "string", "type": "string",
"description": "The PowerShell script to execute." "description": "The PowerShell script to execute."
} }
}, },
@@ -608,9 +608,9 @@ def _gemini_tool_declaration() -> Optional[types.Tool]:
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({
"name": spec["name"], "name": spec["name"],
"description": spec["description"], "description": spec["description"],
"parameters": spec["parameters"] "parameters": spec["parameters"]
}) })
if _agent_tools.get(TOOL_NAME, True): if _agent_tools.get(TOOL_NAME, True):
raw_tools.append({ raw_tools.append({
@@ -626,7 +626,7 @@ def _gemini_tool_declaration() -> Optional[types.Tool]:
"type": "object", "type": "object",
"properties": { "properties": {
"script": { "script": {
"type": "string", "type": "string",
"description": "The PowerShell script to execute." "description": "The PowerShell script to execute."
} }
}, },
@@ -637,22 +637,22 @@ def _gemini_tool_declaration() -> Optional[types.Tool]:
_BIAS_ENGINE.apply_semantic_nudges(raw_tools, _active_tool_preset) _BIAS_ENGINE.apply_semantic_nudges(raw_tools, _active_tool_preset)
declarations: list[types.FunctionDeclaration] = [] declarations: list[types.FunctionDeclaration] = []
for tool_def in raw_tools: for tool_def in raw_tools:
props = {} props = {}
params = tool_def.get("parameters", {}) params = tool_def.get("parameters", {})
for pname, pdef in params.get("properties", {}).items(): for pname, pdef in params.get("properties", {}).items():
ptype_str = pdef.get("type", "string").upper() ptype_str = pdef.get("type", "string").upper()
ptype = getattr(types.Type, ptype_str, types.Type.STRING) ptype = getattr(types.Type, ptype_str, types.Type.STRING)
props[pname] = types.Schema( props[pname] = types.Schema(
type=ptype, type=ptype,
description=pdef.get("description", ""), description=pdef.get("description", ""),
) )
declarations.append(types.FunctionDeclaration( declarations.append(types.FunctionDeclaration(
name=tool_def["name"], name = tool_def["name"],
description=tool_def["description"], description = tool_def["description"],
parameters=types.Schema( parameters = types.Schema(
type=types.Type.OBJECT, type = types.Type.OBJECT,
properties=props, properties = props,
required=params.get("required", []), required = params.get("required", []),
), ),
)) ))
return types.Tool(function_declarations=declarations) if declarations else None return types.Tool(function_declarations=declarations) if declarations else None
@@ -662,13 +662,13 @@ def _gemini_tool_declaration() -> Optional[types.Tool]:
#region: Tool Execution #region: Tool Execution
async def _execute_tool_calls_concurrently( async def _execute_tool_calls_concurrently(
calls: list[Any], calls: list[Any],
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]]],
qa_callback: Optional[Callable[[str], str]], qa_callback: Optional[Callable[[str], str]],
r_idx: int, r_idx: int,
provider: str, provider: str,
patch_callback: Optional[Callable[[str, str], Optional[str]]] = None patch_callback: Optional[Callable[[str, str], Optional[str]]] = None
) -> list[tuple[str, str, str, str]]: # tool_name, call_id, output, original_name ) -> list[tuple[str, str, str, str]]: # tool_name, call_id, output, original_name
""" """
Executes tool calls concurrently using asyncio.gather. Executes tool calls concurrently using asyncio.gather.
@@ -702,32 +702,29 @@ async def _execute_tool_calls_concurrently(
""" """
monitor = performance_monitor.get_monitor() monitor = performance_monitor.get_monitor()
if monitor.enabled: monitor.start_component("ai_client._execute_tool_calls_concurrently") if monitor.enabled: monitor.start_component("ai_client._execute_tool_calls_concurrently")
tier = get_current_tier() tier = get_current_tier()
tasks = [] tasks = []
for fc in calls: for fc in calls:
if provider == "gemini": 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
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": elif provider == "anthropic": name, args, call_id = cast(str, getattr(fc, "name")), cast(dict[str, Any], getattr(fc, "input")), cast(str, getattr(fc, "id"))
name, args, call_id = cast(str, fc.get("name")), cast(dict[str, Any], fc.get("args", {})), cast(str, fc.get("id")) elif provider == "deepseek":
elif provider == "anthropic": tool_info = fc.get("function", {})
name, args, call_id = cast(str, getattr(fc, "name")), cast(dict[str, Any], getattr(fc, "input")), cast(str, getattr(fc, "id")) name = cast(str, tool_info.get("name"))
elif provider == "deepseek":
tool_info = fc.get("function", {})
name = cast(str, tool_info.get("name"))
tool_args_str = cast(str, tool_info.get("arguments", "{}")) tool_args_str = cast(str, tool_info.get("arguments", "{}"))
call_id = cast(str, fc.get("id")) call_id = cast(str, fc.get("id"))
try: args = json.loads(tool_args_str) try: args = json.loads(tool_args_str)
except: args = {} except: args = {}
elif provider == "minimax": elif provider == "minimax":
tool_info = fc.get("function", {}) tool_info = fc.get("function", {})
name = cast(str, tool_info.get("name")) name = cast(str, tool_info.get("name"))
tool_args_str = cast(str, tool_info.get("arguments", "{}")) tool_args_str = cast(str, tool_info.get("arguments", "{}"))
call_id = cast(str, fc.get("id")) call_id = cast(str, fc.get("id"))
try: args = json.loads(tool_args_str) try: args = json.loads(tool_args_str)
except: args = {} except: args = {}
else: else:
continue continue
tasks.append(_execute_single_tool_call_async(name, args, call_id, base_dir, pre_tool_callback, qa_callback, r_idx, tier, patch_callback)) tasks.append(_execute_single_tool_call_async(name, args, call_id, base_dir, pre_tool_callback, qa_callback, r_idx, tier, patch_callback))
results = await asyncio.gather(*tasks) results = await asyncio.gather(*tasks)
@@ -735,22 +732,22 @@ async def _execute_tool_calls_concurrently(
return results return results
def run_with_tool_loop( def run_with_tool_loop(
client: Any, client: Any,
request: Union[OpenAICompatibleRequest, Callable[[int], OpenAICompatibleRequest]], request: Union[OpenAICompatibleRequest, Callable[[int], OpenAICompatibleRequest]],
*, *,
capabilities: Optional[VendorCapabilities] = None, capabilities: Optional[VendorCapabilities] = None,
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,
stream_callback: Optional[Callable[[str], None]] = None, stream_callback: Optional[Callable[[str], None]] = None,
patch_callback: Optional[Callable[[str, str], Optional[str]]] = None, patch_callback: Optional[Callable[[str, str], Optional[str]]] = None,
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[list[dict[str, Any]]] = None,
trim_func: Optional[Callable[[list[dict[str, Any]]], None]] = None, trim_func: Optional[Callable[[list[dict[str, Any]]], 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[dict[str, Any]]], list[dict[str, Any]]]] = None,
) -> str: ) -> str:
""" """
Orchestrates the LLM conversation loop, executing tool calls and updating history. Orchestrates the LLM conversation loop, executing tool calls and updating history.
@@ -800,28 +797,23 @@ def run_with_tool_loop(
raise RuntimeError(res.errors[0].message if res.errors else "Unknown OpenAI error") raise RuntimeError(res.errors[0].message if res.errors else "Unknown OpenAI error")
return res.data return res.data
request_builder: Callable[[int], OpenAICompatibleRequest] = (request if callable(request) else (lambda _i: request)) request_builder: Callable[[int], OpenAICompatibleRequest] = (request if callable(request) else (lambda _i: request))
dispatch_send: Callable[[int], NormalizedResponse] = send_func or _default_send dispatch_send: Callable[[int], NormalizedResponse] = send_func or _default_send
response_text: str = "" response_text: str = ""
for _round_idx in range(MAX_TOOL_ROUNDS + 2): for _round_idx in range(MAX_TOOL_ROUNDS + 2):
response = dispatch_send(_round_idx) response = dispatch_send(_round_idx)
reasoning_content: str = reasoning_extractor(response.raw_response) if reasoning_extractor else "" reasoning_content: str = reasoning_extractor(response.raw_response) if reasoning_extractor else ""
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: dict[str, Any] = {"role": "assistant", "content": response.text or None}
if reasoning_content: if reasoning_content: msg["reasoning_content"] = 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)
if not response.tool_calls: if not response.tool_calls: break
break if on_pre_dispatch is not None: _adjusted_calls = on_pre_dispatch(_round_idx, response.tool_calls)
if on_pre_dispatch is not None: else: _adjusted_calls = response.tool_calls
_adjusted_calls = on_pre_dispatch(_round_idx, response.tool_calls)
else:
_adjusted_calls = response.tool_calls
try: try:
loop = asyncio.get_running_loop() loop = asyncio.get_running_loop()
results = asyncio.run_coroutine_threadsafe( results = asyncio.run_coroutine_threadsafe(
_execute_tool_calls_concurrently( _execute_tool_calls_concurrently(
_adjusted_calls, base_dir, pre_tool_callback, qa_callback, _round_idx, vendor_name, patch_callback, _adjusted_calls, base_dir, pre_tool_callback, qa_callback, _round_idx, vendor_name, patch_callback,
@@ -836,24 +828,23 @@ def run_with_tool_loop(
with history_lock: with history_lock:
for _i, (tool_name, call_id, out, _err) in enumerate(results): for _i, (tool_name, call_id, out, _err) in enumerate(results):
history.append({ history.append({
"role": "tool", "role": "tool",
"tool_call_id": call_id, "tool_call_id": call_id,
"content": str(out) if out else "", "content": str(out) if out else "",
}) })
if trim_func is not None: if trim_func is not None: trim_func(history)
trim_func(history)
return response_text return response_text
async def _execute_single_tool_call_async( async def _execute_single_tool_call_async(
name: str, name: str,
args: dict[str, Any], args: dict[str, Any],
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]]],
qa_callback: Optional[Callable[[str], str]], qa_callback: Optional[Callable[[str], str]],
r_idx: int, r_idx: int,
tier: str | None = None, tier: str | None = None,
patch_callback: Optional[Callable[[str, str], Optional[str]]] = None patch_callback: Optional[Callable[[str, str], Optional[str]]] = None
) -> tuple[str, str, str, str]: ) -> tuple[str, str, str, str]:
""" """
Executes a single tool call asynchronously, checking the approval clutch. Executes a single tool call asynchronously, checking the approval clutch.
@@ -889,9 +880,9 @@ async def _execute_single_tool_call_async(
(like pre_tool_callback and _run_script) to separate worker threads using asyncio.to_thread. (like pre_tool_callback and _run_script) to separate worker threads using asyncio.to_thread.
""" """
set_current_tier(tier) set_current_tier(tier)
out = "" out = ""
tool_executed = False tool_executed = False
events.emit("tool_execution", payload={"status": "started", "tool": name, "args": args, "round": r_idx}) events.emit("tool_execution", payload = {"status": "started", "tool": name, "args": args, "round": r_idx})
# Check for auto approval mode # Check for auto approval mode
approval_mode = _tool_approval_modes.get(name, "ask") approval_mode = _tool_approval_modes.get(name, "ask")
@@ -906,24 +897,22 @@ async def _execute_single_tool_call_async(
elif pre_tool_callback: elif pre_tool_callback:
# pre_tool_callback is synchronous and might block for HITL # pre_tool_callback is synchronous and might block for HITL
res = await asyncio.to_thread(pre_tool_callback, scr, base_dir, qa_callback) res = await asyncio.to_thread(pre_tool_callback, scr, base_dir, qa_callback)
if res is None: if res is None: out = "USER REJECTED: tool execution cancelled"
out = "USER REJECTED: tool execution cancelled" else: out = res
else:
out = res
tool_executed = True tool_executed = True
if not tool_executed: if not tool_executed:
is_native = name in mcp_client.TOOL_NAMES is_native = name in mcp_client.TOOL_NAMES
ext_tools = mcp_client.get_external_mcp_manager().get_all_tools() ext_tools = mcp_client.get_external_mcp_manager().get_all_tools()
is_external = name in ext_tools is_external = name in ext_tools
if name and (is_native or is_external): if name and (is_native or is_external):
_append_comms("OUT", "tool_call", {"name": name, "id": call_id, "args": args}) _append_comms("OUT", "tool_call", {"name": name, "id": call_id, "args": args})
should_approve = (name in mcp_client.MUTATING_TOOLS or is_external) and approval_mode != "auto" and pre_tool_callback should_approve = (name in mcp_client.MUTATING_TOOLS or is_external) and approval_mode != "auto" and pre_tool_callback
if should_approve: if should_approve:
label = "MCP MUTATING" if is_native else "EXTERNAL MCP" label = "MCP MUTATING" if is_native else "EXTERNAL MCP"
desc = f"# {label} TOOL: {name}\n" + "\n".join(f"# {k}: {repr(v)}" for k, v in args.items()) desc = f"# {label} TOOL: {name}\n" + "\n".join(f"# {k}: {repr(v)}" for k, v in args.items())
_res = await asyncio.to_thread(pre_tool_callback, desc, base_dir, qa_callback) _res = await asyncio.to_thread(pre_tool_callback, desc, base_dir, qa_callback)
out = "USER REJECTED: tool execution cancelled" if _res is None else await mcp_client.async_dispatch(name, args) out = "USER REJECTED: tool execution cancelled" if _res is None else await mcp_client.async_dispatch(name, args)
else: else:
out = await mcp_client.async_dispatch(name, args) out = await mcp_client.async_dispatch(name, args)
if tool_log_callback: if tool_log_callback:
@@ -936,19 +925,16 @@ async def _execute_single_tool_call_async(
out = f"ERROR: unknown tool '{name}'" out = f"ERROR: unknown tool '{name}'"
if tool_log_callback: if tool_log_callback:
tool_log_callback(f"ERROR: {name}", out) tool_log_callback(f"ERROR: {name}", out)
return (name, call_id, out, name) return (name, call_id, out, name)
def _run_script(script: str, base_dir: str, qa_callback: Optional[Callable[[str], str]] = None, patch_callback: Optional[Callable[[str, str], Optional[str]]] = None) -> str: def _run_script(script: str, base_dir: str, qa_callback: Optional[Callable[[str], str]] = None, patch_callback: Optional[Callable[[str, str], Optional[str]]] = None) -> str:
if confirm_and_run_callback is None: if confirm_and_run_callback is None:
return "ERROR: no confirmation handler registered" return "ERROR: no confirmation handler registered"
result = confirm_and_run_callback(script, base_dir, qa_callback, patch_callback) result = confirm_and_run_callback(script, base_dir, qa_callback, patch_callback)
if result is None: if result is None: output = "USER REJECTED: command was not executed"
output = "USER REJECTED: command was not executed" else: output = result
else: if tool_log_callback is not None: tool_log_callback(script, output)
output = result
if tool_log_callback is not None:
tool_log_callback(script, output)
return output return output
def _truncate_tool_output(output: str) -> str: def _truncate_tool_output(output: str) -> str:
@@ -963,30 +949,25 @@ def _truncate_tool_output(output: str) -> str:
def _reread_file_items(file_items: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: def _reread_file_items(file_items: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
""" """
Re-reads file items from the filesystem if their modification times have changed. Re-reads file items from the filesystem if their modification times have changed.
Functional Purpose: Functional Purpose:
Iterates through context files, compares current filesystem mtime against cached mtime, Iterates through context files, compares current filesystem mtime against cached mtime,
and reads file contents if changes are detected, returning both the full refreshed set and reads file contents if changes are detected, returning both the full refreshed set
and the subset of changed items. and the subset of changed items.
Parameters & Inputs: Parameters & Inputs: file_items (list[dict[str, Any]]): List of file dictionaries containing keys "path" and optionally "mtime", "content".
file_items (list[dict[str, Any]]): List of file dictionaries containing keys "path" and optionally "mtime", "content".
Returns: Returns: tuple[list[dict[str, Any]], list[dict[str, Any]]]: A tuple containing (refreshed_items, changed_items).
tuple[list[dict[str, Any]], list[dict[str, Any]]]: A tuple containing (refreshed_items, changed_items).
Immediate-Mode DAG / Thread Context: Immediate-Mode DAG / Thread Context:
Called by: _send_gemini Called by: _send_gemini
Calls: pathlib.Path.stat, pathlib.Path.read_text Calls: pathlib.Path.stat, pathlib.Path.read_text
SSDL: SSDL: `o-> [I:get_mtime] -> [B:changed?] -> [I:read_file] -> [T:diff_text]`
`o-> [I:get_mtime] -> [B:changed?] -> [I:read_file] -> [T:diff_text]`
Thread Boundaries: Thread Boundaries: Runs synchronously in the caller thread. Does synchronous blocking file system I/O.
Runs synchronously in the caller thread. Does synchronous blocking file system I/O.
""" """
refreshed: list[dict[str, Any]] = [] refreshed: list[dict[str, Any]] = []
changed: list[dict[str, Any]] = [] changed: list[dict[str, Any]] = []
for item in file_items: for item in file_items:
path = item.get("path") path = item.get("path")
if path is None: if path is None:
@@ -995,11 +976,11 @@ def _reread_file_items(file_items: list[dict[str, Any]]) -> tuple[list[dict[str,
p = path if isinstance(path, _P) else _P(path) p = path if isinstance(path, _P) else _P(path)
try: try:
current_mtime = p.stat().st_mtime current_mtime = p.stat().st_mtime
prev_mtime = cast(float, item.get("mtime", 0.0)) prev_mtime = cast(float, item.get("mtime", 0.0))
if current_mtime == prev_mtime: if current_mtime == prev_mtime:
refreshed.append(item) refreshed.append(item)
continue continue
content = p.read_text(encoding="utf-8") content = p.read_text(encoding="utf-8")
new_item = {**item, "old_content": item.get("content", ""), "content": content, "error": False, "mtime": current_mtime} new_item = {**item, "old_content": item.get("content", ""), "content": content, "error": False, "mtime": current_mtime}
refreshed.append(new_item) refreshed.append(new_item)
changed.append(new_item) changed.append(new_item)
@@ -1014,8 +995,8 @@ def _build_file_context_text(file_items: list[dict[str, Any]]) -> str:
return "" return ""
parts: list[str] = [] parts: list[str] = []
for item in file_items: for item in file_items:
path = item.get("path") or item.get("entry", "unknown") path = item.get("path") or item.get("entry", "unknown")
suffix = str(path).rsplit(".", 1)[-1] if "." in str(path) else "text" suffix = str(path).rsplit(".", 1)[-1] if "." in str(path) else "text"
content = item.get("content", "") content = item.get("content", "")
parts.append(f"### `{path}`\n\n```{suffix}\n{content}\n```") parts.append(f"### `{path}`\n\n```{suffix}\n{content}\n```")
return "\n\n---\n\n".join(parts) return "\n\n---\n\n".join(parts)
@@ -1050,34 +1031,29 @@ def _build_file_diff_text(changed_items: list[dict[str, Any]]) -> str:
return "" return ""
parts: list[str] = [] parts: list[str] = []
for item in changed_items: for item in changed_items:
path = item.get("path") or item.get("entry", "unknown") path = item.get("path") or item.get("entry", "unknown")
content = cast(str, item.get("content", "")) content = cast(str, item.get("content", ""))
old_content = cast(str, item.get("old_content", "")) old_content = cast(str, item.get("old_content", ""))
new_lines = content.splitlines(keepends=True) new_lines = content.splitlines(keepends=True)
if len(new_lines) <= _DIFF_LINE_THRESHOLD or not old_content: if len(new_lines) <= _DIFF_LINE_THRESHOLD or not old_content:
suffix = str(path).rsplit(".", 1)[-1] if "." in str(path) else "text" suffix = str(path).rsplit(".", 1)[-1] if "." in str(path) else "text"
parts.append(f"### `{path}` (full)\n\n```{suffix}\n{content}\n```") parts.append(f"### `{path}` (full)\n\n```{suffix}\n{content}\n```")
else: else:
old_lines = old_content.splitlines(keepends=True) old_lines = old_content.splitlines(keepends=True)
diff = difflib.unified_diff(old_lines, new_lines, fromfile=str(path), tofile=str(path), lineterm="") diff = difflib.unified_diff(old_lines, new_lines, fromfile=str(path), tofile=str(path), lineterm="")
diff_text = "\n".join(diff) diff_text = "\n".join(diff)
if diff_text: if diff_text: parts.append(f"### `{path}` (diff)\n\n```diff\n{diff_text}\n```")
parts.append(f"### `{path}` (diff)\n\n```diff\n{diff_text}\n```") 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[dict[str, Any]]:
"""
[C: tests/test_tool_access_exclusion.py:test_build_deepseek_tools_excludes_disabled]
"""
raw_tools: list[dict[str, Any]] = [] raw_tools: list[dict[str, Any]] = []
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({
"name": spec["name"], "name": spec["name"],
"description": spec["description"], "description": spec["description"],
"parameters": spec["parameters"] "parameters": spec["parameters"]
}) })
if _agent_tools.get(TOOL_NAME, True): if _agent_tools.get(TOOL_NAME, True):
raw_tools.append({ raw_tools.append({
@@ -1093,7 +1069,7 @@ def _build_deepseek_tools() -> list[dict[str, Any]]:
"type": "object", "type": "object",
"properties": { "properties": {
"script": { "script": {
"type": "string", "type": "string",
"description": "The PowerShell script to execute." "description": "The PowerShell script to execute."
} }
}, },
@@ -1107,9 +1083,9 @@ def _build_deepseek_tools() -> list[dict[str, Any]]:
tools_list.append({ tools_list.append({
"type": "function", "type": "function",
"function": { "function": {
"name": tool_def["name"], "name": tool_def["name"],
"description": tool_def["description"], "description": tool_def["description"],
"parameters": tool_def["parameters"], "parameters": tool_def["parameters"],
} }
}) })
return tools_list return tools_list
@@ -1123,35 +1099,29 @@ def _get_deepseek_tools() -> list[dict[str, Any]]:
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) -> dict[str, Any]:
if isinstance(block, dict): if isinstance(block, dict): return block
return block if hasattr(block, "model_dump"): return cast(dict[str, Any], block.model_dump())
if hasattr(block, "model_dump"): if hasattr(block, "to_dict"): return cast(dict[str, Any], block.to_dict())
return cast(dict[str, Any], block.model_dump())
if hasattr(block, "to_dict"):
return cast(dict[str, Any], block.to_dict())
block_type = getattr(block, "type", None) block_type = getattr(block, "type", None)
if block_type == "text": if block_type == "text": return {"type": "text", "text": block.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")}
return {"type": "text", "text": str(block)} return {"type": "text", "text": str(block)}
#endregion: File Context Building #endregion: File Context Building
#region: Token Estimation #region: Token Estimation
_CHARS_PER_TOKEN: float = 3.5 _CHARS_PER_TOKEN: float = 3.5
_ANTHROPIC_MAX_PROMPT_TOKENS: int = 180_000 _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: dict[str, Any]) -> int:
cached = msg.get("_est_tokens") cached = msg.get("_est_tokens")
if cached is not None: if cached is not None: return cast(int, cached)
return cast(int, cached)
total_chars = 0 total_chars = 0
content = msg.get("content", "") content = msg.get("content", "")
if isinstance(content, str): if isinstance(content, str):
total_chars += len(content) total_chars += len(content)
elif isinstance(content, list): elif isinstance(content, list):
for block in content: for block in content:
@@ -1174,7 +1144,7 @@ def _invalidate_token_estimate(msg: dict[str, Any]) -> None:
def _estimate_prompt_tokens(system_blocks: list[dict[str, Any]], history: list[dict[str, Any]]) -> int: def _estimate_prompt_tokens(system_blocks: list[dict[str, Any]], history: list[dict[str, Any]]) -> 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", ""))
total += max(1, int(len(text) / _CHARS_PER_TOKEN)) total += max(1, int(len(text) / _CHARS_PER_TOKEN))
total += 2500 total += 2500
for msg in history: for msg in history:
@@ -1207,9 +1177,6 @@ def _strip_stale_file_refreshes(history: list[dict[str, Any]]) -> None:
_invalidate_token_estimate(msg) _invalidate_token_estimate(msg)
def _chunk_text(text: str, chunk_size: int) -> list[str]: def _chunk_text(text: str, chunk_size: int) -> list[str]:
"""
[C: src/rag_engine.py:RAGEngine._chunk_code, src/rag_engine.py:RAGEngine.index_file]
"""
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[dict[str, Any]]:
@@ -1232,10 +1199,9 @@ def _strip_cache_controls(history: list[dict[str, Any]]) -> None:
def _add_history_cache_breakpoint(history: list[dict[str, Any]]) -> None: def _add_history_cache_breakpoint(history: list[dict[str, Any]]) -> 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: if len(user_indices) < 2: return
return
target_idx = user_indices[-2] target_idx = user_indices[-2]
content = history[target_idx].get("content") content = history[target_idx].get("content")
if isinstance(content, list) and content: if isinstance(content, list) and content:
last_block = content[-1] last_block = content[-1]
if isinstance(last_block, dict): if isinstance(last_block, dict):
@@ -1252,11 +1218,10 @@ def _add_history_cache_breakpoint(history: list[dict[str, Any]]) -> None:
def _list_anthropic_models() -> list[str]: def _list_anthropic_models() -> list[str]:
try: try:
anthropic = _require_warmed("anthropic") anthropic = _require_warmed("anthropic")
creds = _load_credentials() creds = _load_credentials()
client = anthropic.Anthropic(api_key=creds["anthropic"]["api_key"]) client = anthropic.Anthropic(api_key=creds["anthropic"]["api_key"])
models: list[str] = [] models: list[str] = []
for m in client.models.list(): for m in client.models.list(): models.append(m.id)
models.append(m.id)
return sorted(models) return sorted(models)
except Exception as exc: except Exception as exc:
raise _classify_anthropic_error(exc) from exc raise _classify_anthropic_error(exc) from exc
@@ -1267,23 +1232,22 @@ def _ensure_anthropic_client() -> None:
if _anthropic_client is None: if _anthropic_client is None:
creds = _load_credentials() creds = _load_credentials()
_anthropic_client = anthropic.Anthropic( _anthropic_client = anthropic.Anthropic(
api_key=creds["anthropic"]["api_key"], api_key = creds["anthropic"]["api_key"],
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[dict[str, Any]], history: list[dict[str, Any]]) -> 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: if est <= _ANTHROPIC_MAX_PROMPT_TOKENS: return 0
return 0
dropped = 0 dropped = 0
while len(history) > 3 and est > _ANTHROPIC_MAX_PROMPT_TOKENS: while len(history) > 3 and est > _ANTHROPIC_MAX_PROMPT_TOKENS:
if history[1].get("role") == "assistant" and len(history) > 2 and history[2].get("role") == "user": if history[1].get("role") == "assistant" and len(history) > 2 and history[2].get("role") == "user":
removed_asst = history.pop(1) removed_asst = history.pop(1)
removed_user = history.pop(1) removed_user = history.pop(1)
dropped += 2 dropped += 2
est -= _estimate_message_tokens(removed_asst) est -= _estimate_message_tokens(removed_asst)
est -= _estimate_message_tokens(removed_user) est -= _estimate_message_tokens(removed_user)
while len(history) > 2 and history[1].get("role") == "assistant" and history[2].get("role") == "user": while len(history) > 2 and history[1].get("role") == "assistant" and history[2].get("role") == "user":
content = history[2].get("content", []) content = history[2].get("content", [])
if isinstance(content, list) and content and isinstance(content[0], dict) and content[0].get("type") == "tool_result": if isinstance(content, list) and content and isinstance(content[0], dict) and content[0].get("type") == "tool_result":
@@ -1295,17 +1259,15 @@ def _trim_anthropic_history(system_blocks: list[dict[str, Any]], history: list[d
else: else:
break break
else: else:
removed = history.pop(1) removed = history.pop(1)
dropped += 1 dropped += 1
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[dict[str, Any]]) -> None:
if not history: if not history: return
return
last = history[-1] last = history[-1]
if last.get("role") != "assistant": if last.get("role") != "assistant": return
return
content = last.get("content", []) content = last.get("content", [])
tool_use_ids: list[str] = [] tool_use_ids: list[str] = []
for block in content: for block in content:
@@ -1326,10 +1288,18 @@ def _repair_anthropic_history(history: list[dict[str, Any]]) -> None:
], ],
}) })
def _send_anthropic(md_content: str, user_message: str, base_dir: str, file_items: list[dict[str, Any]] | 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, stream_callback: Optional[Callable[[str], None]] = None, patch_callback: Optional[Callable[[str, str], Optional[str]]] = None) -> Result[str]: def _send_anthropic(
md_content: str,
user_message: str,
base_dir: str,
file_items: list[dict[str, Any]] | 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,
stream_callback: Optional[Callable[[str], None]] = None,
patch_callback: Optional[Callable[[str, str], Optional[str]]] = None
) -> Result[str]:
""" """
[C: src/ai_server.py:_handle_send]
Functional Purpose: Functional Purpose:
Sends requests to Anthropic models, managing conversation history, prompt caching, token limits, and executing tool loops. Sends requests to Anthropic models, managing conversation history, prompt caching, token limits, and executing tool loops.
Parameters & Inputs: Parameters & Inputs:
@@ -1344,18 +1314,18 @@ def _send_anthropic(md_content: str, user_message: str, base_dir: str, file_item
Runs on whichever thread calls send (typically an async worker thread). Runs on whichever thread calls send (typically an async worker thread).
""" """
anthropic = _require_warmed("anthropic") anthropic = _require_warmed("anthropic")
genai = _require_warmed("google.genai") genai = _require_warmed("google.genai")
types = genai.types types = genai.types
monitor = performance_monitor.get_monitor() monitor = performance_monitor.get_monitor()
if monitor.enabled: monitor.start_component("ai_client._send_anthropic") if monitor.enabled: monitor.start_component("ai_client._send_anthropic")
try: try:
_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[dict[str, Any]] = [{"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[dict[str, Any]] = [{"type": "text", "text": f"[DISCUSSION HISTORY]\n\n{discussion_history}\n\n---\n\n{user_message}"}]
else: else:
@@ -1369,21 +1339,20 @@ def _send_anthropic(md_content: str, user_message: str, base_dir: str, file_item
if _history_trunc_limit > 0 and isinstance(t_content, str) and len(t_content) > _history_trunc_limit: if _history_trunc_limit > 0 and isinstance(t_content, str) and len(t_content) > _history_trunc_limit:
block["content"] = t_content[:_history_trunc_limit] + "\n\n... [TRUNCATED BY SYSTEM TO SAVE TOKENS. Original output was too large.]" block["content"] = t_content[:_history_trunc_limit] + "\n\n... [TRUNCATED BY SYSTEM TO SAVE TOKENS. Original output was too large.]"
modified = True modified = True
if modified: if modified: _invalidate_token_estimate(msg)
_invalidate_token_estimate(msg)
_strip_cache_controls(_anthropic_history) _strip_cache_controls(_anthropic_history)
_repair_anthropic_history(_anthropic_history) _repair_anthropic_history(_anthropic_history)
_anthropic_history.append({"role": "user", "content": user_content}) _anthropic_history.append({"role": "user", "content": user_content})
_add_history_cache_breakpoint(_anthropic_history) _add_history_cache_breakpoint(_anthropic_history)
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[dict[str, Any]]) -> list[dict[str, Any]]:
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):
response: Any = None response: Any = None
dropped = _trim_anthropic_history(system_blocks, _anthropic_history) dropped = _trim_anthropic_history(system_blocks, _anthropic_history)
if dropped > 0: if dropped > 0:
est_tokens = _estimate_prompt_tokens(system_blocks, _anthropic_history) est_tokens = _estimate_prompt_tokens(system_blocks, _anthropic_history)
_append_comms("OUT", "request", { _append_comms("OUT", "request", {
@@ -1392,18 +1361,18 @@ def _send_anthropic(md_content: str, user_message: str, base_dir: str, file_item
f"Estimated {est_tokens} tokens remaining. {len(_anthropic_history)} messages in history.]" f"Estimated {est_tokens} tokens remaining. {len(_anthropic_history)} messages in history.]"
), ),
}) })
events.emit("request_start", payload={"provider": "anthropic", "model": _model, "round": round_idx}) events.emit("request_start", payload={"provider": "anthropic", "model": _model, "round": round_idx})
assert _anthropic_client is not None assert _anthropic_client is not None
if stream_callback: if stream_callback:
with _anthropic_client.messages.stream( with _anthropic_client.messages.stream(
model=_model, model = _model,
max_tokens=_max_tokens, max_tokens = _max_tokens,
temperature=_temperature, temperature = _temperature,
top_p=_top_p, top_p = _top_p,
system=cast(Iterable[anthropic.types.TextBlockParam], system_blocks), system = cast(Iterable[anthropic.types.TextBlockParam], system_blocks),
tools=cast(Iterable[anthropic.types.ToolParam], _get_anthropic_tools()), tools = cast(Iterable[anthropic.types.ToolParam], _get_anthropic_tools()),
messages=cast(Iterable[anthropic.types.MessageParam], _strip_private_keys(_anthropic_history)), messages = cast(Iterable[anthropic.types.MessageParam], _strip_private_keys(_anthropic_history)),
) as stream: ) as stream:
for event in stream: for event in stream:
if isinstance(event, anthropic.types.ContentBlockDeltaEvent) and event.delta.type == "text_delta": if isinstance(event, anthropic.types.ContentBlockDeltaEvent) and event.delta.type == "text_delta":
@@ -1411,17 +1380,17 @@ def _send_anthropic(md_content: str, user_message: str, base_dir: str, file_item
response = stream.get_final_message() response = stream.get_final_message()
else: else:
response = _anthropic_client.messages.create( response = _anthropic_client.messages.create(
model=_model, model = _model,
max_tokens=_max_tokens, max_tokens = _max_tokens,
temperature=_temperature, temperature = _temperature,
top_p=_top_p, top_p = _top_p,
system=cast(Iterable[anthropic.types.TextBlockParam], system_blocks), system = cast(Iterable[anthropic.types.TextBlockParam], system_blocks),
tools=cast(Iterable[anthropic.types.ToolParam], _get_anthropic_tools()), tools = cast(Iterable[anthropic.types.ToolParam], _get_anthropic_tools()),
messages=cast(Iterable[anthropic.types.MessageParam], _strip_private_keys(_anthropic_history)), messages = cast(Iterable[anthropic.types.MessageParam], _strip_private_keys(_anthropic_history)),
) )
serialised_content = [_content_block_to_dict(b) for b in response.content] serialised_content = [_content_block_to_dict(b) for b in response.content]
_anthropic_history.append({ _anthropic_history.append({
"role": "assistant", "role": "assistant",
"content": serialised_content, "content": serialised_content,
}) })
text_blocks = [b.text for b in response.content if hasattr(b, "text") and b.text] text_blocks = [b.text for b in response.content if hasattr(b, "text") and b.text]
@@ -1436,12 +1405,10 @@ def _send_anthropic(md_content: str, user_message: str, base_dir: str, file_item
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
cache_creation = getattr(response.usage, "cache_creation_input_tokens", None) cache_creation = getattr(response.usage, "cache_creation_input_tokens", None)
cache_read = getattr(response.usage, "cache_read_input_tokens", None) cache_read = getattr(response.usage, "cache_read_input_tokens", None)
if cache_creation is not None: if cache_creation is not None: usage_dict["cache_creation_input_tokens"] = cache_creation
usage_dict["cache_creation_input_tokens"] = cache_creation if cache_read is not None: usage_dict["cache_read_input_tokens"] = cache_read
if cache_read is not None:
usage_dict["cache_read_input_tokens"] = cache_read
events.emit("response_received", payload={"provider": "anthropic", "model": _model, "usage": usage_dict, "round": round_idx}) events.emit("response_received", payload={"provider": "anthropic", "model": _model, "usage": usage_dict, "round": round_idx})
_append_comms("IN", "response", { _append_comms("IN", "response", {
"round": round_idx, "round": round_idx,
@@ -1450,21 +1417,19 @@ def _send_anthropic(md_content: str, user_message: str, base_dir: str, file_item
"tool_calls": tool_use_blocks, "tool_calls": tool_use_blocks,
"usage": usage_dict, "usage": usage_dict,
}) })
if response.stop_reason != "tool_use" or not tool_use_blocks: if response.stop_reason != "tool_use" or not tool_use_blocks: break
break if round_idx > MAX_TOOL_ROUNDS: break
if round_idx > MAX_TOOL_ROUNDS:
break
# Execute tools concurrently # Execute tools concurrently
try: try:
loop = asyncio.get_running_loop() loop = asyncio.get_running_loop()
results = asyncio.run_coroutine_threadsafe( results = asyncio.run_coroutine_threadsafe(
_execute_tool_calls_concurrently(response.content, base_dir, pre_tool_callback, qa_callback, round_idx, "anthropic", patch_callback), _execute_tool_calls_concurrently(response.content, base_dir, pre_tool_callback, qa_callback, round_idx, "anthropic", patch_callback),
loop loop
).result() ).result()
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[dict[str, Any]] = []
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)
@@ -1476,7 +1441,7 @@ def _send_anthropic(md_content: str, user_message: str, base_dir: str, file_item
}) })
_append_comms("IN", "tool_result", {"name": name, "id": call_id, "output": out}) _append_comms("IN", "tool_result", {"name": name, "id": call_id, "output": out})
events.emit("tool_execution", payload={"status": "completed", "tool": name, "result": out, "round": round_idx}) events.emit("tool_execution", payload={"status": "completed", "tool": name, "result": out, "round": round_idx})
if _cumulative_tool_bytes > _MAX_TOOL_OUTPUT_BYTES: if _cumulative_tool_bytes > _MAX_TOOL_OUTPUT_BYTES:
tool_results.append({ tool_results.append({
"type": "text", "type": "text",
@@ -1485,7 +1450,7 @@ def _send_anthropic(md_content: str, user_message: str, base_dir: str, file_item
_append_comms("OUT", "request", {"message": f"[TOOL OUTPUT BUDGET EXCEEDED: {_cumulative_tool_bytes} bytes]"}) _append_comms("OUT", "request", {"message": f"[TOOL OUTPUT BUDGET EXCEEDED: {_cumulative_tool_bytes} bytes]"})
if file_items: if file_items:
file_items, changed = _reread_file_items(file_items) file_items, changed = _reread_file_items(file_items)
refreshed_ctx = _build_file_diff_text(changed) refreshed_ctx = _build_file_diff_text(changed)
if refreshed_ctx: if refreshed_ctx:
tool_results.append({ tool_results.append({
"type": "text", "type": "text",
@@ -1510,7 +1475,7 @@ def _send_anthropic(md_content: str, user_message: str, base_dir: str, file_item
], ],
}) })
final_text = "\n\n".join(all_text_parts) final_text = "\n\n".join(all_text_parts)
res = final_text if final_text.strip() else "(No text returned by the model)" res = final_text if final_text.strip() else "(No text returned by the model)"
if monitor.enabled: monitor.end_component("ai_client._send_anthropic") if monitor.enabled: monitor.end_component("ai_client._send_anthropic")
return Result(data=res) return Result(data=res)
except Exception as exc: except Exception as exc:
@@ -1522,19 +1487,15 @@ def _send_anthropic(md_content: str, user_message: str, base_dir: str, file_item
#region: Gemini Provider #region: Gemini Provider
def get_gemini_cache_stats() -> dict[str, Any]: def get_gemini_cache_stats() -> dict[str, Any]:
"""
[C: src/app_controller.py:AppController._recalculate_session_usage, src/app_controller.py:AppController._update_cached_stats, tests/test_ai_cache_tracking.py:test_gemini_cache_tracking, tests/test_gemini_metrics.py:test_get_gemini_cache_stats_with_mock_client]
"""
_ensure_gemini_client() _ensure_gemini_client()
if not _gemini_client: if not _gemini_client: return {"cache_count": 0, "total_size_bytes": 0, "cached_files": []}
return {"cache_count": 0, "total_size_bytes": 0, "cached_files": []} caches_iterator = _gemini_client.caches.list()
caches_iterator = _gemini_client.caches.list() caches = list(caches_iterator)
caches = list(caches_iterator)
total_size_bytes = sum(getattr(c, 'size_bytes', 0) for c in caches) total_size_bytes = sum(getattr(c, 'size_bytes', 0) for c in caches)
return { return {
"cache_count": len(caches), "cache_count": len(caches),
"total_size_bytes": total_size_bytes, "total_size_bytes": total_size_bytes,
"cached_files": _gemini_cached_file_paths, "cached_files": _gemini_cached_file_paths,
} }
def _list_gemini_cli_models() -> list[str]: def _list_gemini_cli_models() -> list[str]:
@@ -1549,98 +1510,90 @@ def _list_gemini_cli_models() -> list[str]:
def _list_gemini_models(api_key: str) -> list[str]: def _list_gemini_models(api_key: str) -> list[str]:
try: try:
genai = _require_warmed("google.genai") genai = _require_warmed("google.genai")
client = genai.Client(api_key=api_key) client = genai.Client(api_key=api_key)
models: list[str] = [] models: list[str] = []
for m in client.models.list(): for m in client.models.list():
name = m.name name = m.name
if name and name.startswith("models/"): if name and name.startswith("models/"): name = name[len("models/"):]
name = name[len("models/"):] if name and "gemini" in name.lower(): models.append(name)
if name and "gemini" in name.lower():
models.append(name)
return sorted(models) return sorted(models)
except Exception as exc: except Exception as exc:
raise _classify_gemini_error(exc) from exc raise _classify_gemini_error(exc) from exc
def _ensure_gemini_client() -> None: def _ensure_gemini_client() -> None:
"""
[C: src/rag_engine.py:GeminiEmbeddingProvider.embed]
"""
global _gemini_client global _gemini_client
genai = _require_warmed("google.genai") genai = _require_warmed("google.genai")
if _gemini_client is None: if _gemini_client is None:
creds = _load_credentials() creds = _load_credentials()
_gemini_client = genai.Client(api_key=creds["gemini"]["api_key"]) _gemini_client = genai.Client(api_key=creds["gemini"]["api_key"])
def _get_gemini_history_list(chat: Any | None) -> list[Any]: def _get_gemini_history_list(chat: Any | None) -> list[Any]:
if not chat: return [] if not chat: return []
if hasattr(chat, "_history"): if hasattr(chat, "_history"): return cast(list[Any], chat._history)
return cast(list[Any], chat._history) if hasattr(chat, "history"): return cast(list[Any], chat.history)
if hasattr(chat, "history"): if hasattr(chat, "get_history"): return cast(list[Any], chat.get_history())
return cast(list[Any], chat.history)
if hasattr(chat, "get_history"):
return cast(list[Any], chat.get_history())
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[dict[str, Any]] | 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,
enable_tools: bool = True, enable_tools: bool = True,
stream_callback: Optional[Callable[[str], None]] = None, stream_callback: Optional[Callable[[str], None]] = None,
patch_callback: Optional[Callable[[str, str], Optional[str]]] = None) -> Result[str]: patch_callback: Optional[Callable[[str, str], Optional[str]]] = None
) -> Result[str]:
""" """
[C: src/ai_server.py:_handle_send, tests/test_tier4_interceptor.py:test_gemini_provider_passes_qa_callback_to_run_script]
Functional Purpose: Sends requests to Gemini via google-genai SDK, handling context caching, chat history, and tools. Functional Purpose: Sends requests to Gemini via google-genai SDK, handling context caching, chat history, and tools.
Parameters & Inputs: md_content, user_message, base_dir, file_items, discussion_history, callbacks, enable_tools. Parameters & Inputs: md_content, user_message, base_dir, file_items, discussion_history, callbacks, enable_tools.
Immediate-Mode DAG / Thread Context: Called by: send; Calls: _ensure_gemini_client, client.caches.create, client.chats.create, run_with_tool_loop Immediate-Mode DAG / Thread Context: Called by: send; Calls: _ensure_gemini_client, client.caches.create, client.chats.create, run_with_tool_loop
SSDL: SSDL: [I:_ensure_gemini_client] -> [B:Cache Changed?] -> [I:client.caches.create] -> [I:client.chats.create] -> [T:Result]
[I:_ensure_gemini_client] -> [B:Cache Changed?] -> [I:client.caches.create] -> [I:client.chats.create] -> [T:Result]
Thread Boundaries: Runs on caller thread (typically an async worker thread). Thread Boundaries: Runs on caller thread (typically an async worker thread).
""" """
global _gemini_chat, _gemini_cache, _gemini_cache_md_hash, _gemini_cache_created_at, _gemini_cached_file_paths global _gemini_chat, _gemini_cache, _gemini_cache_md_hash, _gemini_cache_created_at, _gemini_cached_file_paths
genai = _require_warmed("google.genai") genai = _require_warmed("google.genai")
types = genai.types types = genai.types
monitor = performance_monitor.get_monitor() monitor = performance_monitor.get_monitor()
if monitor.enabled: monitor.start_component("ai_client._send_gemini") if monitor.enabled: monitor.start_component("ai_client._send_gemini")
try: try:
_ensure_gemini_client(); mcp_client.configure(file_items or [], [base_dir]) _ensure_gemini_client(); 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>"
td = _gemini_tool_declaration() if enable_tools else None td = _gemini_tool_declaration() if enable_tools else None
tools_decl = [td] if td else None tools_decl = [td] if td else None
current_md_hash = hashlib.md5(md_content.encode()).hexdigest() current_md_hash = hashlib.md5(md_content.encode()).hexdigest()
old_history = None old_history = None
assert _gemini_client is not None assert _gemini_client is not None
if _gemini_chat and _gemini_cache_md_hash != current_md_hash: if _gemini_chat and _gemini_cache_md_hash != current_md_hash:
old_history = list(_get_gemini_history_list(_gemini_chat)) if _get_gemini_history_list(_gemini_chat) else [] old_history = list(_get_gemini_history_list(_gemini_chat)) if _get_gemini_history_list(_gemini_chat) else []
if _gemini_cache: if _gemini_cache:
try: _gemini_client.caches.delete(name=_gemini_cache.name) try: _gemini_client.caches.delete(name=_gemini_cache.name)
except Exception as e: _append_comms("OUT", "request", {"message": f"[CACHE DELETE WARN] {e}"}) except Exception as e: _append_comms("OUT", "request", {"message": f"[CACHE DELETE WARN] {e}"})
_gemini_chat = None _gemini_chat = None
_gemini_cache = None _gemini_cache = None
_gemini_cache_created_at = None _gemini_cache_created_at = None
_gemini_cached_file_paths = [] _gemini_cached_file_paths = []
_append_comms("OUT", "request", {"message": "[CONTEXT CHANGED] Rebuilding cache and chat session..."}) _append_comms("OUT", "request", {"message": "[CONTEXT CHANGED] Rebuilding cache and chat session..."})
if _gemini_chat and _gemini_cache and _gemini_cache_created_at: if _gemini_chat and _gemini_cache and _gemini_cache_created_at:
elapsed = time.time() - _gemini_cache_created_at elapsed = time.time() - _gemini_cache_created_at
if elapsed > _GEMINI_CACHE_TTL * 0.9: if elapsed > _GEMINI_CACHE_TTL * 0.9:
old_history = list(_get_gemini_history_list(_gemini_chat)) if _get_gemini_history_list(_gemini_chat) else [] old_history = list(_get_gemini_history_list(_gemini_chat)) if _get_gemini_history_list(_gemini_chat) else []
#TODO(Ed): Review(Exception)
try: _gemini_client.caches.delete(name=_gemini_cache.name) try: _gemini_client.caches.delete(name=_gemini_cache.name)
except Exception as e: _append_comms("OUT", "request", {"message": f"[CACHE DELETE WARN] {e}"}) except Exception as e: _append_comms("OUT", "request", {"message": f"[CACHE DELETE WARN] {e}"})
_gemini_chat = None _gemini_chat = None
_gemini_cache = None _gemini_cache = None
_gemini_cache_created_at = None _gemini_cache_created_at = None
_gemini_cached_file_paths = [] _gemini_cached_file_paths = []
_append_comms("OUT", "request", {"message": f"[CACHE TTL] Rebuilding cache (expired after {int(elapsed)}s)..."}) _append_comms("OUT", "request", {"message": f"[CACHE TTL] Rebuilding cache (expired after {int(elapsed)}s)..."})
if not _gemini_chat: if not _gemini_chat:
chat_config = types.GenerateContentConfig( chat_config = types.GenerateContentConfig(
system_instruction=sys_instr, system_instruction = sys_instr,
tools=cast(Any, tools_decl), tools = cast(Any, tools_decl),
temperature=_temperature, temperature = _temperature,
top_p=_top_p, top_p = _top_p,
max_output_tokens=_max_tokens, max_output_tokens = _max_tokens,
safety_settings=[types.SafetySetting(category=types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT, threshold=types.HarmBlockThreshold.BLOCK_ONLY_HIGH)] safety_settings = [types.SafetySetting(category=types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT, threshold=types.HarmBlockThreshold.BLOCK_ONLY_HIGH)]
) )
should_cache = False should_cache = False
try: try: