import copy import json import asyncio import inspect import os import re import time from types import SimpleNamespace from typing import Any, Callable, List, Sequence, Union, Dict, Optional, Mapping from openai.types.chat import ChatCompletionToolParam from openspace.grounding.core.types import ToolSchema, ToolResult, ToolStatus from openspace.grounding.core.tool import BaseTool from openspace.services.conversation.messages import ( build_assistant_api_error_message, build_system_api_error_message, normalize_messages_for_api, ) from openspace.services.conversation.content_blocks import normalize_multimodal_messages_for_model from openspace.services.tooling.prompt import ( ToolPromptContext, tool_to_openai_schema, ) # strip_meta removed in step 7.2 — normalize_messages_for_api supersedes it. from .types import ModelResponse, TokenUsage, token_usage_from_dict from .effort import ( EffortConfig, EffortLevel, build_effort_request_params, get_effort_config, ) from .thinking import ( ThinkingConfig, build_thinking_request_params, get_model_max_output_tokens, get_thinking_config, ) from openspace.utils.logging import Logger from .errors import ( CannotRetryError, FallbackTriggeredError, OverloadedError, RetryContext, FLOOR_OUTPUT_TOKENS, MAX_OVERLOADED_RETRIES, get_default_max_retries, classify_api_error, get_retry_after_ms, get_retry_delay, is_model_not_available_error as _is_model_not_available, is_overloaded_error as _is_overloaded, is_prompt_too_long_error as _is_prompt_too_long, parse_max_tokens_context_overflow_error, should_retry as _should_retry, to_model_not_available_error, to_prompt_too_long_error, ) # .env loading is centralized in host_detection.resolver.load_runtime_env(). # CLI/MCP entrypoints call it before reading startup env vars, and the # resolver helpers also call it defensively. logger = Logger.get_logger(__name__) def _usage_has_counts(usage: TokenUsage) -> bool: return any( int(getattr(usage, attr, 0) or 0) > 0 for attr in ( "input_tokens", "output_tokens", "total_tokens", "cache_read_input_tokens", "cache_creation_input_tokens", "reasoning_tokens", ) ) or float(getattr(usage, "cost", 0.0) or 0.0) > 0 def _estimate_missing_usage( *, model: str, input_messages: Sequence[Mapping[str, Any]], output_text: str, ) -> TokenUsage: try: from openspace.services.conversation.compact import estimate_message_tokens input_tokens = int(estimate_message_tokens(input_messages, model=model)) output_tokens = int( estimate_message_tokens( [{"role": "assistant", "content": output_text}], model=model, ) ) except Exception: input_tokens = max(1, len(json.dumps(list(input_messages), ensure_ascii=False).encode("utf-8")) // 4) output_tokens = max(1, len(output_text.encode("utf-8")) // 4) output_tokens = max(1 if output_text else 0, output_tokens) return TokenUsage( input_tokens=max(0, input_tokens), output_tokens=output_tokens, total_tokens=max(0, input_tokens) + output_tokens, ) def _env_bool(name: str, default: bool = False) -> bool: raw = os.environ.get(name) if raw is None: return default return raw.strip().lower() in {"1", "true", "yes", "y", "on"} def _is_official_deepseek_model(model: str) -> bool: normalized = str(model or "").strip().lower() return normalized.startswith("deepseek/") or normalized.startswith("deepseek-") def _supports_required_tool_choice( model: str, *, deepseek_thinking_disabled: bool = False, ) -> bool: """Return whether the provider accepts API-level required tool choice.""" if _env_bool("OPENSPACE_ALLOW_DEEPSEEK_REQUIRED_TOOL_CHOICE", False): return True if _is_official_deepseek_model(model) and not deepseek_thinking_disabled: return False return True def _deepseek_thinking_mode_from_env() -> str | None: raw = ( os.environ.get("OPENSPACE_DEEPSEEK_THINKING") or os.environ.get("OPENSPACE_DEEPSEEK_THINKING_MODE") ) if raw is None: return None normalized = raw.strip().lower().replace("_", "-") aliases = { "": None, "auto": None, "default": None, "enabled": "enabled", "enable": "enabled", "on": "enabled", "thinking": "enabled", "disabled": "disabled", "disable": "disabled", "off": "disabled", "non-thinking": "disabled", "nonthinking": "disabled", } if normalized not in aliases: logger.warning( "Ignoring invalid OPENSPACE_DEEPSEEK_THINKING value: %s", raw, ) return aliases.get(normalized) def _deepseek_request_params(model: str, *, tool_choice: str) -> dict[str, Any]: """Build DeepSeek-specific request controls when explicitly configured.""" if not _is_official_deepseek_model(model): return {} mode = _deepseek_thinking_mode_from_env() if ( mode is None and tool_choice == "required" and _env_bool( "OPENSPACE_DEEPSEEK_DISABLE_THINKING_ON_REQUIRED_TOOL_CHOICE", False, ) ): mode = "disabled" params: dict[str, Any] = {} if mode in {"enabled", "disabled"}: params["extra_body"] = {"thinking": {"type": mode}} effort = os.environ.get("OPENSPACE_DEEPSEEK_REASONING_EFFORT") if effort and effort.strip() and mode != "disabled": params["reasoning_effort"] = effort.strip().lower() return params def _deepseek_thinking_disabled(params: Mapping[str, Any]) -> bool: extra_body = params.get("extra_body") if not isinstance(extra_body, Mapping): return False thinking = extra_body.get("thinking") return ( isinstance(thinking, Mapping) and str(thinking.get("type") or "").strip().lower() == "disabled" ) def _disable_reasoning_on_required_tool_choice(model: str) -> bool: """Return whether required-tool recovery calls should suppress reasoning.""" if _env_bool("OPENSPACE_DISABLE_REASONING_ON_REQUIRED_TOOL_CHOICE", False): return True normalized = str(model or "").strip().lower() if normalized.startswith("openrouter/"): return _env_bool( "OPENSPACE_OPENROUTER_DISABLE_REASONING_ON_REQUIRED_TOOL_CHOICE", False, ) return False def _required_tool_choice_reasoning_disabled_params(model: str) -> dict[str, Any]: """Provider-specific params that actively disable reasoning for tool recovery.""" normalized = str(model or "").strip().lower() if normalized.startswith("openrouter/"): return {"reasoning": {"effort": "none", "exclude": True}} return {} def _decode_jsonish_string(value: str) -> str: """Decode the small escape subset models commonly use in malformed JSON.""" return ( value.replace("\\r\\n", "\n") .replace("\\n", "\n") .replace("\\r", "\r") .replace("\\t", "\t") .replace('\\"', '"') .replace("\\/", "/") .replace("\\\\", "\\") ) def _extract_jsonish_string_field( raw: str, field_names: Sequence[str], *, greedy: bool = False, ) -> str | None: """Extract a JSON string field from malformed model-emitted arguments.""" for field_name in field_names: match = re.search(rf'"{re.escape(field_name)}"\s*:\s*"', raw) if not match: continue start = match.end() if greedy: end = raw.rfind('"') if end > start: return _decode_jsonish_string(raw[start:end]) return None index = start escaped = False while index < len(raw): char = raw[index] if escaped: escaped = False elif char == "\\": escaped = True elif char == '"': return _decode_jsonish_string(raw[start:index]) index += 1 return None def _salvage_malformed_write_arguments( raw_arguments: str, tool_name: str, ) -> dict[str, Any] | None: """Recover common huge ``write`` calls whose JSON string was not escaped.""" if tool_name != "write": return None file_path = _extract_jsonish_string_field( raw_arguments, ("file_path", "path", "filename"), ) content = _extract_jsonish_string_field( raw_arguments, ("content",), greedy=True, ) if not file_path or content is None: return None return {"file_path": file_path, "content": content} def _text_tool_call( *, name: str, arguments: Mapping[str, Any], index: int = 0, ) -> Dict[str, Any]: return { "id": f"call_text_{index}", "type": "function", "function": { "name": name, "arguments": json.dumps(dict(arguments), ensure_ascii=False), }, } def _parse_text_tool_calls( content: str, tool_map: Mapping[str, BaseTool], ) -> list[Dict[str, Any]]: if not _env_bool("OPENSPACE_PARSE_TEXT_TOOL_CALLS") or not content: return [] function_match = re.search( r"(.*?)", content, flags=re.DOTALL, ) if function_match: tool_name = function_match.group(1).strip() body = function_match.group(2) if tool_name in tool_map: arguments: dict[str, Any] = {} for param_match in re.finditer( r"(.*?)", body, flags=re.DOTALL, ): arguments[param_match.group(1).strip()] = param_match.group(2).strip() if arguments: return [_text_tool_call(name=tool_name, arguments=arguments)] json_blocks = re.findall( r"```(?:json)?\s*(\{.*?\})\s*```", content, flags=re.DOTALL | re.IGNORECASE, ) stripped = content.strip() if not json_blocks and stripped.startswith("{") and stripped.endswith("}"): json_blocks = [stripped] for block in json_blocks: try: payload = json.loads(block) except json.JSONDecodeError: continue if not isinstance(payload, dict): continue command = payload.get("command") if isinstance(command, str) and command.strip() and "bash" in tool_map: arguments = {"command": command.strip()} description = payload.get("description") if isinstance(description, str) and description.strip(): arguments["description"] = description.strip() return [_text_tool_call(name="bash", arguments=arguments)] name = payload.get("name") or payload.get("tool") arguments = payload.get("arguments") or payload.get("parameters") if isinstance(name, str) and name in tool_map and isinstance(arguments, dict): return [_text_tool_call(name=name, arguments=arguments)] return [] def _merge_completion_kwargs( target: Dict[str, Any], updates: Mapping[str, Any], ) -> None: """Merge LiteLLM kwargs, preserving nested ``extra_body`` values.""" for key, value in updates.items(): if ( key == "extra_body" and isinstance(value, Mapping) and isinstance(target.get("extra_body"), Mapping) ): merged = dict(target["extra_body"]) for nested_key, nested_value in value.items(): if ( isinstance(nested_value, Mapping) and isinstance(merged.get(nested_key), Mapping) ): merged[nested_key] = { **dict(merged[nested_key]), **dict(nested_value), } else: merged[nested_key] = nested_value target[key] = merged else: target[key] = value def _ensure_stream_usage_options(target: Dict[str, Any]) -> None: """Ask OpenAI-compatible streaming providers to include final usage stats.""" stream_options = target.get("stream_options") if stream_options is None: target["stream_options"] = {"include_usage": True} return if isinstance(stream_options, Mapping): options = dict(stream_options) options.setdefault("include_usage", True) target["stream_options"] = options def _serializable_effort_value(value: Any) -> str | int | None: if isinstance(value, EffortLevel): return value.value if isinstance(value, int): return value if value is None: return None return str(value) def _find_tool_for_api_normalization( tool_map: Mapping[str, BaseTool], name: str, ) -> BaseTool | None: tool = tool_map.get(name) if tool is not None: return tool for candidate in tool_map.values(): aliases = getattr(candidate, "aliases", ()) or () if getattr(candidate, "name", None) == name or name in aliases: return candidate return None def _normalize_tool_inputs_for_api( messages: Sequence[Mapping[str, Any]], tool_map: Mapping[str, BaseTool], ) -> list[dict[str, Any]]: if not tool_map: return [dict(message) for message in messages] from openspace.tool_runtime.pipeline.execution import normalize_tool_input_for_api normalized_messages: list[dict[str, Any]] = [] for message in messages: copied_message = dict(message) tool_calls = copied_message.get("tool_calls") if copied_message.get("role") != "assistant" or not isinstance(tool_calls, list): normalized_messages.append(copied_message) continue normalized_tool_calls: list[Any] = [] for tool_call in tool_calls: if not isinstance(tool_call, Mapping): normalized_tool_calls.append(tool_call) continue function = tool_call.get("function") if not isinstance(function, Mapping): normalized_tool_calls.append(dict(tool_call)) continue tool_name = str(function.get("name") or "") tool = _find_tool_for_api_normalization(tool_map, tool_name) if tool is None: normalized_tool_calls.append(dict(tool_call)) continue raw_arguments = function.get("arguments") arguments_were_string = isinstance(raw_arguments, str) if arguments_were_string: try: parsed_arguments = json.loads(raw_arguments or "{}") except (TypeError, json.JSONDecodeError): normalized_tool_calls.append(dict(tool_call)) continue else: parsed_arguments = raw_arguments if not isinstance(parsed_arguments, dict): normalized_tool_calls.append(dict(tool_call)) continue normalized_arguments = normalize_tool_input_for_api(tool, parsed_arguments) copied_function = dict(function) copied_function["arguments"] = ( json.dumps(normalized_arguments, ensure_ascii=False) if arguments_were_string else normalized_arguments ) normalized_tool_calls.append({ **dict(tool_call), "function": copied_function, }) copied_message["tool_calls"] = normalized_tool_calls normalized_messages.append(copied_message) return normalized_messages _litellm_module: Any | None = None def _get_litellm() -> Any: global _litellm_module if _litellm_module is None: import litellm as imported_litellm # Disable LiteLLM verbose logging to prevent stdout blocking with large # tool schemas. Do this lazily so importing LLMClient does not trigger # LiteLLM/httpx proxy initialization in tests that never call a model. imported_litellm.set_verbose = False imported_litellm.suppress_debug_info = True _litellm_module = imported_litellm return _litellm_module class _LiteLLMProxy: def __getattr__(self, name: str) -> Any: return getattr(_get_litellm(), name) litellm = _LiteLLMProxy() def _last_user_text(messages: Sequence[Mapping[str, Any]]) -> str | None: for msg in reversed(list(messages)): if msg.get("role") != "user": continue content = msg.get("content") if isinstance(content, str): return content if isinstance(content, list): parts: list[str] = [] for block in content: if isinstance(block, Mapping): text = block.get("text") or block.get("content") if isinstance(text, str): parts.append(text) elif isinstance(block, str): parts.append(block) return "\n".join(parts) if parts else None return None def _sanitize_schema(params: Dict) -> Dict: """Sanitize tool parameter schema to comply with Claude API requirements.""" if not params: return {"type": "object", "properties": {}, "required": []} # Deep copy to avoid modifying the original import copy sanitized = copy.deepcopy(params) # Anthropic API requires top-level type to be 'object' # If it's not an object, wrap the schema as a property of an object top_level_type = sanitized.get("type") if top_level_type and top_level_type != "object": # Wrap non-object schema as a single property called "value" logger.debug(f"[SCHEMA_SANITIZE] Wrapping non-object schema (type={top_level_type}) into object") wrapped = { "type": "object", "properties": { "value": sanitized # The original schema becomes a property }, "required": ["value"] # Make it required } sanitized = wrapped # If type is object but missing properties/required, add them if sanitized.get("type") == "object": if "properties" not in sanitized: sanitized["properties"] = {} if "required" not in sanitized: sanitized["required"] = [] # Remove non-standard fields that may cause issues (like 'title') sanitized.pop("title", None) # Recursively sanitize nested properties if "properties" in sanitized and isinstance(sanitized["properties"], dict): for prop_name, prop_schema in list(sanitized["properties"].items()): if isinstance(prop_schema, dict): # Remove title from nested properties prop_schema.pop("title", None) return sanitized def _schema_to_openai(schema: ToolSchema) -> ChatCompletionToolParam: """Convert ToolSchema to OpenAI ChatCompletion tool format""" function_def = { "name": schema.name, "description": schema.description or "", } # Sanitize and add parameters if schema.parameters: sanitized = _sanitize_schema(schema.parameters) function_def["parameters"] = sanitized # Debug: verify sanitization worked if "title" in schema.parameters and "title" not in sanitized: logger.debug(f"Sanitized tool '{schema.name}': removed title") else: # Claude requires parameters field even if empty function_def["parameters"] = {"type": "object", "properties": {}, "required": []} return { "type": "function", "function": function_def } async def _prepare_tools_for_llmclient( tools: List[BaseTool] | None, fmt: str = "openai", *, prompt_context: ToolPromptContext | Any | None = None, use_schema_cache: bool = True, ) -> tuple[Sequence[Union[ToolSchema, ChatCompletionToolParam]], Dict[str, BaseTool]]: """Convert BaseTool list to LLMClient usable format, with deduplication. Args: tools: BaseTool instance list (should be obtained from GroundingClient and bound to runtime_info) if None or empty list, return empty list fmt: output format, "openai" for OpenAI format """ if not tools: return [], {} if fmt == "openai": result = [] tool_map = {} # llm_name -> BaseTool name_count = {} for tool in tools: name = tool.schema.name name_count[name] = name_count.get(name, 0) + 1 seen_names = set() for tool in tools: original_name = tool.schema.name if name_count[original_name] > 1: server_name = "unknown" if tool.is_bound and tool.runtime_info and tool.runtime_info.server_name: server_name = tool.runtime_info.server_name llm_name = f"{server_name}__{original_name}" else: llm_name = original_name if llm_name in seen_names: logger.warning(f"[TOOL_DEDUP] Skipping duplicate tool: {llm_name}") continue seen_names.add(llm_name) tool_param = await tool_to_openai_schema( tool, llm_name=llm_name, prompt_context=prompt_context, sanitize_schema=_sanitize_schema, use_cache=use_schema_cache, ) result.append(tool_param) tool_map[llm_name] = tool if llm_name != original_name: logger.info(f"[TOOL_RENAME] {original_name} -> {llm_name}") logger.info(f"[SCHEMA_SANITIZE] Prepared {len(result)} tools for LLM (from {len(tools)} total)") return result, tool_map tool_map = {tool.schema.name: tool for tool in tools} return [tool.schema for tool in tools], tool_map # ═══════════════════════════════════════════════════════════════════════ # Legacy helper functions removed in step 7.2. # # The following module-level functions were deleted because they are # superseded by the new tool execution pipeline (steps 5.1 / 5.2): # # _infer_backend_from_tool_name → removed with the old LLM+tool wrapper # _resolve_tool_call_target → find_tool_by_name() in tool_execution.py # _summarize_tool_result → OpenSpace uses persist-to-disk, not LLM summary # _tool_result_to_message_async → pipeline step 7 (build_tool_result_message) # _execute_tool_call → run_tool_use() in tool_execution.py # # Old callers should use: # - call_model() for pure LLM calls # - run_tool_use() / run_tools() for tool execution # - enforce_tool_result_budget() for large result handling # ═══════════════════════════════════════════════════════════════════════ class LLMClient: """LLMClient class for single round call""" def __init__( self, model: str = "openrouter/anthropic/claude-sonnet-4.5", enable_thinking: bool = False, rate_limit_delay: float = 0.0, max_retries: int | None = None, retry_delay: float = 1.0, timeout: float = 120.0, fallback_model: Optional[str] = None, **litellm_kwargs, ): """ Args: model: LLM model identifier. enable_thinking: Whether to enable extended thinking mode. rate_limit_delay: Minimum delay between API calls (0 = no delay). max_retries: Maximum retry attempts on rate-limit / transient errors. retry_delay: Initial retry delay in seconds (exponential backoff). timeout: Request timeout in seconds (default: 120 s). fallback_model: Consecutive-overloaded fallback model (``None`` = no fallback). **litellm_kwargs: Additional litellm parameters. """ self.model = model self.enable_thinking = enable_thinking self.rate_limit_delay = rate_limit_delay self.max_retries = max_retries if max_retries is not None else get_default_max_retries() self.retry_delay = retry_delay self.timeout = timeout self.fallback_model = fallback_model self.litellm_kwargs = litellm_kwargs self._logger = Logger.get_logger(__name__) self._last_call_time = 0.0 self._event_callback: Optional[Callable[..., Any]] = None self._usage_callback: Optional[Callable[[str, TokenUsage], Any]] = None def set_event_callback(self, callback: Optional[Callable[..., Any]]) -> None: """Set callback for streaming events. Signature: async callback(event_type: str, data: dict)""" self._event_callback = callback def set_usage_callback( self, callback: Optional[Callable[[str, TokenUsage], Any]], ) -> None: """Set callback invoked after each model usage update. Signature: ``callback(model, usage: TokenUsage)``. """ self._usage_callback = callback async def _emit_event(self, event_type: str, data: Dict[str, Any]) -> None: """Emit an event via callback if set. Failures are silently swallowed.""" if self._event_callback is None: return try: await self._event_callback(event_type, data) except Exception: pass async def _invoke_callback(self, callback: Optional[Callable[..., Any]], *args, **kwargs) -> Any: """Invoke a sync or async callback and return its result.""" if callback is None: return None result = callback(*args, **kwargs) if inspect.isawaitable(result): return await result return result async def _invoke_usage_callback(self, model: str, usage: TokenUsage) -> None: if self._usage_callback is None: return await self._invoke_callback(self._usage_callback, model, usage) async def _emit_text_tokens(self, text: str, chunk_size: int = 64) -> None: """Emit coarse-grained text chunks when true provider token streaming is unavailable.""" if not text: return for idx in range(0, len(text), chunk_size): await self._emit_event("llm_token", {"token": text[idx: idx + chunk_size]}) @staticmethod def _response_field(value: Any, name: str, default: Any = None) -> Any: if isinstance(value, Mapping): return value.get(name, default) return getattr(value, name, default) @staticmethod def _stringify_stream_tool_arguments(value: Any) -> str: if value is None: return "" if isinstance(value, str): return value if isinstance(value, (Mapping, list, tuple)): return json.dumps(value, ensure_ascii=False) return str(value) @classmethod def _merge_stream_tool_call_deltas( cls, tool_calls_by_index: dict[int, dict[str, Any]], delta_tool_calls: Any, ) -> None: if not delta_tool_calls: return for fallback_index, delta_tool_call in enumerate(delta_tool_calls): raw_index = cls._response_field( delta_tool_call, "index", fallback_index, ) try: index = int(raw_index) except (TypeError, ValueError): index = fallback_index accumulated = tool_calls_by_index.setdefault( index, { "id": "", "type": "function", "function": {"name": "", "arguments": ""}, }, ) tool_call_id = cls._response_field(delta_tool_call, "id", None) if tool_call_id: accumulated["id"] = str(tool_call_id) tool_call_type = cls._response_field(delta_tool_call, "type", None) if tool_call_type: accumulated["type"] = str(tool_call_type) function_delta = cls._response_field(delta_tool_call, "function", None) if not function_delta: continue name_delta = cls._response_field(function_delta, "name", None) if name_delta: accumulated["function"]["name"] = str(name_delta) arguments_delta = cls._response_field(function_delta, "arguments", None) if arguments_delta: accumulated["function"]["arguments"] += ( cls._stringify_stream_tool_arguments(arguments_delta) ) @classmethod def _stream_tool_call_objects( cls, tool_calls_by_index: Mapping[int, Mapping[str, Any]], ) -> list[Any]: tool_calls: list[Any] = [] for index in sorted(tool_calls_by_index): tool_call = tool_calls_by_index[index] function = tool_call.get("function") or {} tool_calls.append( SimpleNamespace( id=str(tool_call.get("id") or f"call_{index}"), type=str(tool_call.get("type") or "function"), function=SimpleNamespace( name=str(function.get("name") or ""), arguments=str(function.get("arguments") or ""), ), ) ) return tool_calls @classmethod def _stream_reasoning_delta(cls, delta: Any) -> Any: reasoning = ( cls._response_field(delta, "reasoning_content", None) or cls._response_field(delta, "reasoning", None) ) if reasoning: return reasoning provider_fields = cls._response_field( delta, "provider_specific_fields", None, ) if provider_fields: return cls._response_field(provider_fields, "reasoning", None) return None async def _consume_streaming_completion( self, stream: Any, *, abort_event: Optional[asyncio.Event] = None, emit_events: bool = True, ) -> Any: content_parts: list[str] = [] reasoning_parts: list[Any] = [] tool_calls_by_index: dict[int, dict[str, Any]] = {} finish_reason: str | None = None raw_usage: Any = None response_model: str | None = None response_id: str | None = None saw_choice = False async for chunk in stream: if abort_event is not None and abort_event.is_set(): raise asyncio.CancelledError("Request was aborted") response_model = response_model or self._response_field( chunk, "model", None, ) response_id = response_id or self._response_field(chunk, "id", None) chunk_usage = self._response_field(chunk, "usage", None) if chunk_usage is not None: raw_usage = chunk_usage for choice in self._response_field(chunk, "choices", []) or []: saw_choice = True choice_finish_reason = self._response_field( choice, "finish_reason", None, ) if choice_finish_reason is not None: finish_reason = choice_finish_reason delta = self._response_field(choice, "delta", None) if delta is None: delta = self._response_field(choice, "message", None) if delta is None: continue content_delta = self._response_field(delta, "content", None) if isinstance(content_delta, str) and content_delta: content_parts.append(content_delta) if emit_events: await self._emit_event( "llm_token", {"token": content_delta}, ) reasoning_delta = self._stream_reasoning_delta(delta) if reasoning_delta: reasoning_parts.append(reasoning_delta) self._merge_stream_tool_call_deltas( tool_calls_by_index, self._response_field(delta, "tool_calls", None), ) message = SimpleNamespace( content="".join(content_parts), tool_calls=self._stream_tool_call_objects(tool_calls_by_index) or None, ) if reasoning_parts: setattr(message, "reasoning_content", "".join(map(str, reasoning_parts))) response = SimpleNamespace( id=response_id, model=response_model, choices=( [ SimpleNamespace( message=message, finish_reason=finish_reason, ) ] if saw_choice else [] ), usage=raw_usage, ) setattr(response, "_openspace_streamed", True) return response # Permission checks are handled by the run_tool_use() pipeline, which # resolves PermissionEngine decisions from the ToolUseContext. Legacy # permission_check PreToolUse adapters may still be registered explicitly by # external callers, but are not part of the default hook set. @staticmethod def _merge_consecutive_system_messages(messages: List[Dict]) -> List[Dict]: """Merge consecutive system messages into one. Providers like MiniMax reject requests that contain multiple consecutive messages with the same role (error 2013 "invalid chat setting"). Merging is safe for all providers — it simply concatenates the content. """ if not messages: return messages merged: List[Dict] = [] for msg in messages: if ( merged and msg.get("role") == "system" and merged[-1].get("role") == "system" ): merged[-1] = { "role": "system", "content": merged[-1].get("content", "") + "\n\n" + msg.get("content", ""), } else: merged.append(msg.copy()) return merged @staticmethod def _is_minimax_model(model: str) -> bool: return isinstance(model, str) and "minimax" in model.lower() @classmethod def _rewrite_nonleading_system_messages_for_minimax( cls, messages: List[Dict], ) -> List[Dict]: """Rewrite non-leading system messages into internal user notes for MiniMax.""" rewritten: List[Dict] = [] rewritten_count = 0 for msg in messages: msg_copy = msg.copy() if msg_copy.get("role") == "system" and rewritten: content = msg_copy.get("content", "") if isinstance(content, str): msg_copy["content"] = ( "[INTERNAL ORCHESTRATION NOTE]\n" "This note was originally injected as a system message by the " "agent runtime. Treat it as workflow guidance, not as a new " "end-user request.\n\n" f"{content}" ) msg_copy["role"] = "user" rewritten_count += 1 rewritten.append(msg_copy) if rewritten_count: logger.info( "Rewrote %d non-leading system message(s) for MiniMax compatibility", rewritten_count, ) return rewritten @classmethod def _normalize_messages_for_model(cls, messages: List[Dict], model: str) -> List[Dict]: """Normalize message history only when a provider requires it.""" normalized = normalize_multimodal_messages_for_model(messages, model) if not cls._is_minimax_model(model): return normalized minimized_system_history = cls._merge_consecutive_system_messages(normalized) return cls._rewrite_nonleading_system_messages_for_minimax( minimized_system_history ) @staticmethod def _serialize_response_field(value): """Convert provider response fields into plain Python containers.""" if hasattr(value, "model_dump"): return value.model_dump(exclude_none=True) if isinstance(value, list): return [LLMClient._serialize_response_field(item) for item in value] if isinstance(value, tuple): return [LLMClient._serialize_response_field(item) for item in value] if isinstance(value, dict): return { key: LLMClient._serialize_response_field(item) for key, item in value.items() } return value async def _rate_limit(self): """Apply rate limiting by adding delay between API calls""" if self.rate_limit_delay > 0: current_time = time.time() time_since_last_call = current_time - self._last_call_time if time_since_last_call < self.rate_limit_delay: sleep_time = self.rate_limit_delay - time_since_last_call self._logger.debug(f"Rate limiting: waiting {sleep_time:.2f}s before next API call") await asyncio.sleep(sleep_time) self._last_call_time = time.time() async def _call_with_retry( self, *, abort_event: Optional[asyncio.Event] = None, fallback_model: Optional[str] = None, thinking_budget_tokens: int = 0, emit_events: bool = True, **completion_kwargs, ): """Call LLM with OpenSpace retry logic.""" request_model = completion_kwargs.get("model", self.model) retry_context = RetryContext( model=request_model, thinking_budget_tokens=max(0, int(thinking_budget_tokens or 0)), ) consecutive_overloaded = 0 last_error: BaseException | None = None max_retries = self.max_retries for attempt in range(1, max_retries + 2): # 1..max_retries+1 inclusive if abort_event is not None and abort_event.is_set(): raise asyncio.CancelledError("Request was aborted") try: response = await asyncio.wait_for( litellm.acompletion(**completion_kwargs), timeout=self.timeout, ) if completion_kwargs.get("stream") and hasattr(response, "__aiter__"): response = await asyncio.wait_for( self._consume_streaming_completion( response, abort_event=abort_event, emit_events=emit_events, ), timeout=self.timeout, ) return response except asyncio.CancelledError: raise # never swallow cancellation except asyncio.TimeoutError as e: last_error = TimeoutError( f"LLM call timed out after {self.timeout}s" ) self._logger.error( "LLM call timed out after %ss (attempt %d/%d)", self.timeout, attempt, max_retries + 1, ) # Timeout is retryable if attempt > max_retries: raise CannotRetryError(last_error, retry_context) from e delay_ms = get_retry_delay(attempt) self._logger.info("Retrying after %d ms...", delay_ms) await asyncio.sleep(delay_ms / 1000) continue except Exception as e: last_error = e self._logger.debug( "API error (attempt %d/%d): %s", attempt, max_retries + 1, str(e)[:300], ) # Non-retryable: prompt too long if _is_prompt_too_long(e): raise to_prompt_too_long_error( e, model=request_model, ) from e # Non-retryable: model / auth not available if _is_model_not_available(e): raise to_model_not_available_error( e, model=request_model, ) from e # Track consecutive overloaded if _is_overloaded(e): consecutive_overloaded += 1 if consecutive_overloaded >= MAX_OVERLOADED_RETRIES: if fallback_model: raise FallbackTriggeredError( request_model, fallback_model, ) from e # No fallback → wrap as CannotRetryError raise CannotRetryError( OverloadedError( str(e), model=request_model, original_error=e, ), retry_context, ) from e else: consecutive_overloaded = 0 if attempt > max_retries: raise CannotRetryError(e, retry_context) from e if not _should_retry(e): raise CannotRetryError(e, retry_context) from e # max_tokens overflow overflow = parse_max_tokens_context_overflow_error(e) if overflow is not None: input_tokens = overflow["input_tokens"] context_limit = overflow["context_limit"] safety_buffer = 1000 available = max(0, context_limit - input_tokens - safety_buffer) if available < FLOOR_OUTPUT_TOKENS: self._logger.error( "Available context %d < FLOOR_OUTPUT_TOKENS %d", available, FLOOR_OUTPUT_TOKENS, ) raise CannotRetryError(e, retry_context) from e min_required = retry_context.thinking_budget_tokens + 1 adjusted = max(FLOOR_OUTPUT_TOKENS, available, min_required) retry_context.max_tokens_override = adjusted self._logger.info( "max_tokens overflow: adjusting to %d (input=%d, limit=%d)", adjusted, input_tokens, context_limit, ) if "max_tokens" in completion_kwargs: completion_kwargs["max_tokens"] = adjusted continue # Compute delay error_tag = classify_api_error(e) retry_after = get_retry_after_ms(e) if retry_after is not None: delay_ms = retry_after else: delay_ms = get_retry_delay(attempt) if error_tag == "rate_limit" and self.rate_limit_delay > 0: delay_ms = max(delay_ms, int(self.rate_limit_delay * 1000)) self._logger.warning( "%s error (attempt %d/%d), waiting %d ms before retry...", error_tag, attempt, max_retries + 1, delay_ms, ) retry_system_msg = build_system_api_error_message( error_msg=str(e)[:500], retry_in_ms=delay_ms, retry_attempt=attempt, max_retries=max_retries, ) if emit_events: await self._emit_event("llm_retry", { "attempt": attempt, "max_retries": max_retries, "delay_ms": delay_ms, "error_type": error_tag, "error_message": str(e)[:500], "system_message": retry_system_msg, }) await asyncio.sleep(delay_ms / 1000) # Should not reach here, but safety net raise CannotRetryError(last_error, retry_context) @staticmethod def _safe_parse_tool_arguments( raw_arguments: str | None, tool_name: str, ) -> dict: """Parse tool call arguments JSON with fallback to empty dict.""" if not raw_arguments: return {} try: parsed = json.loads(raw_arguments) if isinstance(parsed, dict): return parsed logger.warning( "[TOOL_JSON] Tool '%s' arguments parsed to %s, using {}", tool_name, type(parsed).__name__, ) return {} except (json.JSONDecodeError, TypeError, ValueError) as exc: logger.warning( "[TOOL_JSON] Failed to parse tool '%s' arguments " "(len=%d): %s. Falling back to {}.", tool_name, len(raw_arguments), str(exc)[:200], ) return {} @staticmethod def _sanitize_tool_arguments_for_history( raw_arguments: str | None, tool_name: str, ) -> tuple[str, dict[str, Any], bool]: """Return API-safe tool arguments and whether the model JSON was invalid.""" if not raw_arguments: return "{}", {}, False try: parsed = json.loads(raw_arguments) if isinstance(parsed, dict): return raw_arguments, parsed, False logger.warning( "[TOOL_JSON] Tool '%s' arguments parsed to %s; " "sanitizing history payload.", tool_name, type(parsed).__name__, ) except (json.JSONDecodeError, TypeError, ValueError) as exc: logger.warning( "[TOOL_JSON] Failed to parse tool '%s' arguments " "(len=%d): %s. Sanitizing history payload.", tool_name, len(raw_arguments), str(exc)[:200], ) salvaged = _salvage_malformed_write_arguments(raw_arguments, tool_name) if salvaged is not None: logger.warning( "[TOOL_JSON] Recovered malformed '%s' arguments " "(len=%d) for history/execution.", tool_name, len(raw_arguments), ) return json.dumps(salvaged, ensure_ascii=False), salvaged, True fallback = { "__openspace_tool_argument_error": ( "The model emitted malformed tool-call JSON. Retry this tool " "call with compact, valid JSON arguments." ), "raw_argument_chars": len(raw_arguments), } return json.dumps(fallback, ensure_ascii=False), fallback, True @staticmethod def _get_error_message_if_refusal( stop_reason: str | None, model: str, ) -> dict | None: """Build an error assistant message when the model refuses.""" if stop_reason not in ("refusal", "content_filter"): return None content = ( "API Error: The model is unable to respond to this request, " "which appears to violate the provider's Usage Policy. " "Try rephrasing the request or attempting a different approach." ) return build_assistant_api_error_message( content, error_details=f"stop_reason={stop_reason}, model={model}" ) @staticmethod def is_api_error_message(message: Dict[str, Any] | None) -> bool: if not isinstance(message, dict): return False meta = message.get("_meta") return isinstance(meta, dict) and bool(meta.get("is_api_error_message")) @classmethod def get_model_response_followup_messages( cls, model_response: ModelResponse, ) -> list[dict[str, Any]]: assistant_index = next( ( index for index, message in enumerate(model_response.messages) if message is model_response.assistant_message ), -1, ) if assistant_index < 0: assistant_role = model_response.assistant_message.get("role") assistant_content = model_response.assistant_message.get("content") assistant_tool_calls = model_response.assistant_message.get("tool_calls") for index in range(len(model_response.messages) - 1, -1, -1): message = model_response.messages[index] if not isinstance(message, dict): continue if message.get("role") != assistant_role: continue if message.get("content") != assistant_content: continue if message.get("tool_calls") != assistant_tool_calls: continue assistant_index = index break if assistant_index < 0: return [] return [ copy.deepcopy(message) for message in model_response.messages[assistant_index + 1 :] if isinstance(message, dict) ] @classmethod def model_response_has_api_error( cls, model_response: ModelResponse, ) -> bool: if cls.is_api_error_message(model_response.assistant_message): return True return any( cls.is_api_error_message(message) for message in cls.get_model_response_followup_messages(model_response) ) async def call_model( self, messages: List[Dict], tools: List[BaseTool] | None = None, **kwargs, ) -> "ModelResponse": """Single-round LLM call — returns ModelResponse, never executes tools. Args: messages: Conversation history (OpenAI format dicts). ``_meta`` fields are stripped before sending to API. tools: Optional BaseTool list. Schemas are converted to OpenAI format and returned in ``ModelResponse.tool_map``. **kwargs: Overrides forwarded to LiteLLM or used internally: - ``model``: override self.model for this call - ``abort_event``: ``asyncio.Event`` to cancel the call - ``fallback_model``: override self.fallback_model - ``enable_thinking``: override self.enable_thinking - ``thinking_config``: explicit ``ThinkingConfig`` or mapping - ``reasoning_effort``: provider effort/depth; ``None`` means auto - ``effort_config``: pre-resolved ``EffortConfig`` - ``tool_choice``: tool selection mode (default ``"auto"``) - ``streaming``: enable provider token streaming (default ``True``) - ``emit_events``: stream LLM UI events (default ``True``) - Any other key is forwarded to ``litellm.acompletion`` Returns: ``ModelResponse`` with ``assistant_message``, ``tool_calls``, ``tool_map``, ``stop_reason``, ``usage``, ``messages``. Raises: PromptTooLongError: Prompt exceeds context window (Layer 1). ModelNotAvailableError: Model/credentials unavailable (Layer 1). CannotRetryError: All retries exhausted (Layer 1). FallbackTriggeredError: Consecutive overloaded → switch model. ValueError: API returned empty response. """ # Extract call_model-specific kwargs request_model: str = kwargs.pop("model", self.model) abort_event = kwargs.pop("abort_event", None) fallback_model: str | None = kwargs.pop( "fallback_model", self.fallback_model, ) enable_thinking: bool = kwargs.pop( "enable_thinking", self.enable_thinking, ) tool_choice: str = kwargs.pop("tool_choice", "auto") reasoning_effort: str | int | None = kwargs.pop("reasoning_effort", None) explicit_effort_config = kwargs.pop("effort_config", None) explicit_thinking_config = kwargs.pop("thinking_config", None) strip_thinking_keep_recent_override = kwargs.pop( "strip_thinking_keep_recent", None, ) tool_prompt_context = kwargs.pop("tool_prompt_context", None) use_tool_schema_cache: bool = bool(kwargs.pop("use_tool_schema_cache", True)) explicit_stream = kwargs.pop("stream", None) explicit_streaming = kwargs.pop("streaming", None) if explicit_streaming is None and "streaming" in self.litellm_kwargs: explicit_streaming = self.litellm_kwargs.get("streaming") if explicit_stream is None and "stream" in self.litellm_kwargs: explicit_stream = self.litellm_kwargs.get("stream") streaming = bool( explicit_streaming if explicit_streaming is not None else True if explicit_stream is None else explicit_stream ) emit_events = bool(kwargs.pop("emit_events", True)) raw_max_tokens = kwargs.pop("max_tokens", None) if raw_max_tokens is None: raw_max_tokens = self.litellm_kwargs.get("max_tokens") try: max_tokens = int(raw_max_tokens) if raw_max_tokens is not None else None except (TypeError, ValueError): max_tokens = None if max_tokens is None or max_tokens <= 0: max_tokens = get_model_max_output_tokens(request_model) # Prepare tools llm_tools: list = [] tool_map: Dict[str, BaseTool] = {} if tools: llm_tools, tool_map = await _prepare_tools_for_llmclient( tools, fmt="openai", prompt_context=tool_prompt_context, use_schema_cache=use_tool_schema_cache, ) disable_reasoning_for_required_tool = ( bool(llm_tools) and tool_choice == "required" and _disable_reasoning_on_required_tool_choice(request_model) ) if disable_reasoning_for_required_tool: enable_thinking = False explicit_effort_config = None explicit_thinking_config = ThinkingConfig.disabled( source="required_tool_choice" ) if isinstance(explicit_effort_config, EffortConfig): effort_config = explicit_effort_config else: effort_config = get_effort_config(request_model, reasoning_effort) effective_effort = _serializable_effort_value(effort_config.applied_value) if ( explicit_thinking_config is None and effort_config.thinking_budget_tokens is not None ): explicit_thinking_config = ThinkingConfig.enabled( effort_config.thinking_budget_tokens, source=effort_config.source, ) thinking_config = get_thinking_config( request_model, effective_effort, user_request=_last_user_text(messages), max_output_tokens=max_tokens, enable_thinking=enable_thinking, explicit=explicit_thinking_config, has_tools=bool(llm_tools), ) thinking_params, thinking_budget_tokens = build_thinking_request_params( thinking_config, request_model, effort=effective_effort, max_output_tokens=max_tokens, ) effort_params = build_effort_request_params(effort_config, request_model) if disable_reasoning_for_required_tool: thinking_params = {} thinking_budget_tokens = 0 effort_params = {} # Build litellm completion kwargs completion_kwargs: Dict[str, Any] = { "model": request_model, **self.litellm_kwargs, "max_tokens": max_tokens, } completion_kwargs.pop("streaming", None) if streaming: completion_kwargs["stream"] = True else: completion_kwargs.pop("stream", None) deepseek_params = _deepseek_request_params( request_model, tool_choice=tool_choice, ) effective_tool_choice = tool_choice if ( llm_tools and tool_choice == "required" and not _supports_required_tool_choice( request_model, deepseek_thinking_disabled=_deepseek_thinking_disabled( deepseek_params ), ) ): effective_tool_choice = "auto" logger.info( "Downgrading tool_choice=required to auto for model %s because " "the provider does not support required tool choice.", request_model, ) if llm_tools: completion_kwargs["tools"] = llm_tools completion_kwargs["tool_choice"] = effective_tool_choice logger.info( "LLM request controls: model=%s tools=%s tool_choice=%s requested_tool_choice=%s " "disable_reasoning_for_required_tool=%s max_tokens=%s", request_model, len(llm_tools), effective_tool_choice if llm_tools else "none", tool_choice, disable_reasoning_for_required_tool, max_tokens, ) _merge_completion_kwargs(completion_kwargs, thinking_params) _merge_completion_kwargs(completion_kwargs, effort_params) _merge_completion_kwargs(completion_kwargs, deepseek_params) # Forward remaining kwargs (e.g. max_tokens, temperature) _merge_completion_kwargs(completion_kwargs, kwargs) if streaming: _ensure_stream_usage_options(completion_kwargs) if disable_reasoning_for_required_tool: completion_kwargs.pop("reasoning", None) completion_kwargs.pop("reasoning_effort", None) completion_kwargs.pop("thinking", None) _merge_completion_kwargs( completion_kwargs, _required_tool_choice_reasoning_disabled_params(request_model), ) if "thinking" in thinking_params: # Anthropic requires temperature=1/default whenever thinking is enabled. completion_kwargs.pop("temperature", None) # Normalize messages # strip_meta → drop compact boundaries → ensure tool_result # pairing → merge consecutive same-role → model-specific fixups. strip_thinking_keep_recent = ( int(strip_thinking_keep_recent_override) if strip_thinking_keep_recent_override is not None else (0 if thinking_config.type == "disabled" else 1) ) api_messages = normalize_messages_for_api( messages, strip_thinking_keep_recent=strip_thinking_keep_recent, ) api_messages = _normalize_tool_inputs_for_api(api_messages, tool_map) api_messages = self._normalize_messages_for_model( api_messages, request_model, ) completion_kwargs["messages"] = api_messages # Rate limit await self._rate_limit() # Emit llm_start event if emit_events: await self._emit_event("llm_start", { "model": request_model, "tool_count": len(llm_tools), "enable_thinking": thinking_config.type != "disabled", "thinking_type": thinking_config.type, "thinking_budget_tokens": thinking_budget_tokens, "thinking_source": thinking_config.source, "effort": effective_effort, "display_effort": effort_config.level.value, "effort_source": effort_config.source, "streaming": streaming, }) # Call LLM via _call_with_retry # Exceptions (PromptTooLongError, ModelNotAvailableError, # CannotRetryError, FallbackTriggeredError) propagate to caller. response = await self._call_with_retry( abort_event=abort_event, fallback_model=fallback_model, thinking_budget_tokens=thinking_budget_tokens, emit_events=emit_events, **completion_kwargs, ) # Validate response if not response.choices: raise ValueError("LLM response has no choices") choice = response.choices[0] response_message = choice.message # Parse usage → TokenUsage raw_usage = getattr(response, "usage", None) if raw_usage is not None: if hasattr(raw_usage, "model_dump"): usage = token_usage_from_dict(raw_usage.model_dump()) elif isinstance(raw_usage, Mapping): usage = token_usage_from_dict(dict(raw_usage)) elif hasattr(raw_usage, "__dict__"): usage = token_usage_from_dict(vars(raw_usage)) else: usage = TokenUsage() else: usage = TokenUsage() # Parse content & stop_reason content_text: str = response_message.content or "" stop_reason: str | None = getattr(choice, "finish_reason", None) if not _usage_has_counts(usage) and content_text: usage = _estimate_missing_usage( model=request_model, input_messages=api_messages, output_text=content_text, ) # Emit tokens & llm_complete streamed = bool(getattr(response, "_openspace_streamed", False)) if emit_events and not streamed: await self._emit_text_tokens(content_text) if emit_events: await self._emit_event("llm_complete", { "model": request_model, "usage": { "input_tokens": usage.input_tokens, "output_tokens": usage.output_tokens, }, "input_tokens": usage.input_tokens, "output_tokens": usage.output_tokens, "content_length": len(content_text), "stop_reason": stop_reason, "streaming": streamed or streaming, }) await self._invoke_usage_callback(request_model, usage) # Build assistant_message (OpenAI format) assistant_message: Dict[str, Any] = { "role": "assistant", "content": content_text, } # Serialize thinking/reasoning fields. reasoning_content = getattr(response_message, "reasoning_content", None) if not reasoning_content: psf = getattr(response_message, "provider_specific_fields", None) if psf and isinstance(psf, dict): reasoning_content = psf.get("reasoning") if reasoning_content: assistant_message["reasoning_content"] = self._serialize_response_field( reasoning_content, ) msg_name = getattr(response_message, "name", None) if msg_name: assistant_message["name"] = msg_name # Keep OpenAI-compatible tool call shape. Arguments remain provider # wire values here; tool_runtime.pipeline.execution owns parse/fallback. tool_calls_raw = getattr(response_message, "tool_calls", None) parsed_tool_calls: list[Dict[str, Any]] = [] malformed_tool_arguments: list[dict[str, Any]] = [] if tool_calls_raw: for tc in tool_calls_raw: tc_name = tc.function.name ( safe_arguments, _parsed_arguments, malformed_arguments, ) = self._sanitize_tool_arguments_for_history( tc.function.arguments, tc_name, ) if malformed_arguments: malformed_tool_arguments.append({ "id": tc.id, "tool": tc_name, "raw_argument_chars": len(tc.function.arguments or ""), }) parsed_tool_calls.append({ "id": tc.id, "type": "function", "function": { "name": tc_name, "arguments": safe_arguments, }, }) if not parsed_tool_calls: parsed_tool_calls = _parse_text_tool_calls(content_text, tool_map) if parsed_tool_calls: logger.info( "Parsed %d text tool call(s) from assistant content", len(parsed_tool_calls), ) if parsed_tool_calls: assistant_message["tool_calls"] = parsed_tool_calls # Attach _meta to assistant_message assistant_message["_meta"] = { "usage": { "input_tokens": usage.input_tokens, "output_tokens": usage.output_tokens, "cache_creation_input_tokens": usage.cache_creation_input_tokens, "cache_read_input_tokens": usage.cache_read_input_tokens, "reasoning_tokens": usage.reasoning_tokens, "total_tokens": usage.total_tokens, "cost": usage.cost, "web_search_requests": usage.web_search_requests, }, "model": request_model, "stop_reason": stop_reason, "timestamp": time.time(), "thinking_type": thinking_config.type, "thinking_budget_tokens": thinking_budget_tokens, "thinking_source": thinking_config.source, "effort": effective_effort, "display_effort": effort_config.level.value, "effort_source": effort_config.source, } if malformed_tool_arguments: assistant_message["_meta"][ "malformed_tool_arguments" ] = malformed_tool_arguments # Build output messages list # messages = normalized input + assistant response (+ optional # error messages). Caller should ONLY append assistant_message # to its own list — see ModelResponse.messages docstring. output_messages: list[Dict] = list(api_messages) output_messages.append(assistant_message) # Implementation: getErrorMessageIfRefusal (errors.ts L1184-1207) refusal_msg = self._get_error_message_if_refusal( stop_reason, request_model, ) if refusal_msg is not None: output_messages.append(refusal_msg) if stop_reason == "length": output_messages.append(build_assistant_api_error_message( "Response was cut off because it reached the maximum " "output token limit. The model may not have completed " "its response.", error_details=f"stop_reason=length, model={request_model}", )) # Return ModelResponse return ModelResponse( assistant_message=assistant_message, tool_calls=parsed_tool_calls, tool_map=tool_map, stop_reason=stop_reason, usage=usage, messages=output_messages, effective_model=request_model, ) async def call_model_with_fallback( self, messages: List[Dict], tools: List[BaseTool] | None = None, **kwargs, ) -> "ModelResponse": """Variant of :meth:`call_model` that transparently applies the consecutive-overloaded fallback switch. ``call_model`` surfaces :class:`FallbackTriggeredError` to the caller so the main agent loop can update its own bookkeeping (event emission, ``stop_reason_final``, TUI sync). Auxiliary call sites (``BaseAgent.get_llm_response``, ``ExecutionAnalyzer``, ``SkillEvolver``, compaction, etc.) don't have that bookkeeping and previously treated the error as a hard failure — meaning the same provider overload that survives under the main loop would abort the auxiliary path. This helper unifies fallback semantics for those callers: 1. Call :meth:`call_model` with the provided arguments. 2. If it raises :class:`FallbackTriggeredError` and the fallback model differs from the currently effective model, retry once using a request-local ``model=`` override. 3. If the fallback model is missing, equal to the current model, or itself raises :class:`FallbackTriggeredError`, the error is propagated — callers can still surface a terminal failure. The shared ``self.model`` and ``self.fallback_model`` defaults are not mutated. Callers that run multiple auxiliary iterations should carry ``ModelResponse.effective_model`` forward as their local model. All other exceptions (``CannotRetryError``, ``PromptTooLongError``, ``ModelNotAvailableError``, abort, …) propagate unchanged. """ try: response = await self.call_model(messages=messages, tools=tools, **kwargs) response.effective_model = str(kwargs.get("model", self.model) or "") return response except FallbackTriggeredError as fb_err: fallback_model = str(fb_err.fallback_model or "").strip() current_model = str(kwargs.get("model", self.model) or "") if not fallback_model or fallback_model == current_model: raise self._logger.warning( "Auxiliary fallback switch: %s -> %s", fb_err.original_model, fallback_model, ) retry_kwargs = dict(kwargs) retry_kwargs["model"] = fallback_model response = await self.call_model( messages=messages, tools=tools, **retry_kwargs, ) response.effective_model = fallback_model return response @staticmethod def _find_seed_tool_use_context( tools: Sequence[BaseTool] | None, ) -> Any | None: """Return the first tool-attached ToolUseContext-like object, if any.""" if not tools: return None for tool in tools: context = getattr(tool, "_current_context", None) if context is not None: return context return None @classmethod def _infer_auxiliary_cwd( cls, tools: Sequence[BaseTool] | None, seed_context: Any | None = None, ) -> str: """Best-effort cwd for auxiliary tool execution.""" seed_cwd = getattr(seed_context, "cwd", None) if isinstance(seed_cwd, str) and seed_cwd.strip(): return seed_cwd if tools: for tool in tools: direct_cwd = getattr(tool, "_default_working_dir", None) if isinstance(direct_cwd, str) and direct_cwd.strip(): return direct_cwd session = getattr(tool, "_session", None) session_cwd = getattr(session, "default_working_dir", None) if isinstance(session_cwd, str) and session_cwd.strip(): return session_cwd tool_cwd = getattr(tool, "default_working_dir", None) if isinstance(tool_cwd, str) and tool_cwd.strip(): return tool_cwd return os.getcwd() @staticmethod def _load_auxiliary_permission_context( cwd: str, mode: str | None, ) -> Any: """Load workspace permission context for auxiliary callers. Mirrors ``GroundingAgent._resolve_permission_context``: aggregates ``.openspace/settings*.json`` sources (userSettings, projectSettings, localSettings) plus runtime stores into a real :class:`ToolPermissionContext`. Loader failures are surfaced instead of replacing the real workspace state with an empty default. """ try: from openspace.grounding.core.permissions.loader import ( load_tool_permission_context, ) return load_tool_permission_context(cwd, mode) except Exception as exc: logger.error( "Failed to load tool permission context for %s (mode=%s)", cwd, mode, exc_info=True, ) raise RuntimeError( "Failed to load tool permission settings for this workspace" ) from exc @staticmethod def _build_default_hook_registry() -> Any: """Create the same built-in hook registry used by the main agent loop.""" from openspace.services.tooling.hooks import HookRegistry, setup_default_hooks registry = HookRegistry() setup_default_hooks(registry) return registry def build_auxiliary_tool_use_context( self, *, tools: Sequence[BaseTool], messages: list[dict[str, Any]], model: str | None = None, cwd: str | None = None, agent_id: str | None = None, agent_type: str | None = None, max_result_size_chars: int | None = None, abort_event: asyncio.Event | None = None, read_file_state: dict[str, Any] | None = None, tool_results_token_count: int | None = None, permission_engine: Any | None = None, permission_mode: str | None = None, permission_context: Any | None = None, hook_registry: Any | None = None, tui_available: bool | None = None, is_async_agent: bool | None = None, event_sink: Any | None = None, recording_manager: Any | None = None, quality_manager: Any | None = None, parent_task_id: str | None = None, task_description: str | None = None, current_iteration: int | None = None, max_iterations: int | None = None, ) -> Any: """Build a ToolUseContext for non-primary loops that execute tools. Preference order: 1. Explicit arguments from the caller. 2. Context already attached to tools from a prior main-loop execution. 3. Reasonable runtime defaults (real cwd, workspace-loaded permission context via :func:`load_tool_permission_context`, event sink from the LLM client). ``tui_available`` deliberately **defaults to ``False`` (fail-closed)** when no explicit value and no seed context attribute is available. Mirrors ``GroundingAgent._resolve_tui_available``: the presence of a generic event callback (``set_event_callback``) or a runtime event sink on a sub-agent is *not* enough — those are one-way streaming hooks, not an interactive TUI that can reply to ``tool_permission_ask``. Entering the ask flow without a real responder otherwise hangs auxiliary callers on the permission ask timeout. """ from openspace.grounding.core.tool.base import DEFAULT_MAX_RESULT_SIZE_CHARS from openspace.services.tooling.context import ( ToolUseContext, clone_skill_invocation_scopes, ) seed_context = self._find_seed_tool_use_context(tools) resolved_cwd = cwd or self._infer_auxiliary_cwd(tools, seed_context) resolved_permission_context = ( permission_context or getattr(seed_context, "permission_context", None) ) # Track whether the caller/seed actually supplied a mode. When they # didn't we must leave ``mode`` unset so the workspace settings loader # can honour ``permissions.defaultMode`` from ``.openspace/settings.json`` # instead of being overridden by a hard-coded fallback. explicit_permission_mode = ( permission_mode or getattr(seed_context, "permission_mode", None) or getattr(resolved_permission_context, "mode", None) ) if resolved_permission_context is None: # Auxiliary loops must honour the same workspace permission cascade # as the primary agent loop. resolved_permission_context = self._load_auxiliary_permission_context( resolved_cwd, explicit_permission_mode, ) resolved_permission_mode = ( explicit_permission_mode or getattr(resolved_permission_context, "mode", None) or "default" ) resolved_max_result_size = max_result_size_chars if resolved_max_result_size is None: resolved_max_result_size = getattr( seed_context, "max_result_size_chars", DEFAULT_MAX_RESULT_SIZE_CHARS, ) resolved_read_file_state = read_file_state if resolved_read_file_state is None: seeded_state = getattr(seed_context, "read_file_state", None) resolved_read_file_state = dict(seeded_state or {}) resolved_tool_results_tokens = tool_results_token_count if resolved_tool_results_tokens is None: resolved_tool_results_tokens = int( getattr(seed_context, "tool_results_token_count", 0) or 0 ) resolved_hook_registry = ( hook_registry or getattr(seed_context, "hook_registry", None) or self._build_default_hook_registry() ) resolved_event_sink = ( event_sink or getattr(seed_context, "event_sink", None) or self._event_callback ) # TUI availability must mirror ``GroundingAgent._resolve_tui_available``: # an *interactive* sink that can round-trip ``tool_permission_ask`` -> # ``tool_permission_response`` is required before we offer the ask # flow. ``_event_callback`` (set via :meth:`set_event_callback`) and # per-agent ``_runtime_event_sink`` wiring are *unidirectional* # streaming hooks — they propagate events outward but cannot resolve # a pending ask. Treating their presence as evidence of a TUI makes # auxiliary callers (BaseAgent, ExecutionAnalyzer, SkillEvolver, ...) # enter the ask flow, emit # ``tool_permission_ask``, and then hang on the 300s ask timeout # because no responder exists. Main loop is fail-closed here; the # auxiliary paths must match that semantics. Callers with a real TUI # (e.g. the public application facade) should pass ``tui_available=True`` explicitly. # or surface it via the seed ``ToolUseContext``. if tui_available is not None: resolved_tui_available = bool(tui_available) elif seed_context is not None and hasattr(seed_context, "tui_available"): resolved_tui_available = bool(getattr(seed_context, "tui_available")) else: resolved_tui_available = False return ToolUseContext( tools=list(tools), all_tools=list(tools), model=str( model or getattr(seed_context, "model", None) or getattr(self, "model", "unknown") ), llm_client=self, cwd=resolved_cwd, original_cwd=str( getattr(seed_context, "original_cwd", None) or resolved_cwd ), agent_id=agent_id or getattr(seed_context, "agent_id", None) or "auxiliary", agent_type=agent_type if agent_type is not None else getattr(seed_context, "agent_type", None), max_result_size_chars=int(resolved_max_result_size), abort_event=abort_event or getattr(seed_context, "abort_event", None), messages=messages, read_file_state=resolved_read_file_state, tool_results_token_count=resolved_tool_results_tokens, permission_engine=permission_engine or getattr(seed_context, "permission_engine", None), permission_mode=resolved_permission_mode, permission_context=resolved_permission_context, hook_registry=resolved_hook_registry, tui_available=resolved_tui_available, is_async_agent=( is_async_agent if is_async_agent is not None else bool(getattr(seed_context, "is_async_agent", False)) ), event_sink=resolved_event_sink, recording_manager=( recording_manager or getattr(seed_context, "recording_manager", None) ), quality_manager=( quality_manager or getattr(seed_context, "quality_manager", None) ), parent_task_id=( parent_task_id if parent_task_id is not None else getattr(seed_context, "parent_task_id", None) ), task_id=getattr(seed_context, "task_id", None), task_description=( task_description if task_description is not None else str(getattr(seed_context, "task_description", "") or "") ), current_iteration=( int(current_iteration) if current_iteration is not None else int(getattr(seed_context, "current_iteration", 0) or 0) ), max_iterations=( int(max_iterations) if max_iterations is not None else int(getattr(seed_context, "max_iterations", 0) or 0) ), session_id=getattr(seed_context, "session_id", None), session_dir=getattr(seed_context, "session_dir", None), tool_results_dir=getattr(seed_context, "tool_results_dir", None), session_storage=getattr(seed_context, "session_storage", None), file_history=getattr(seed_context, "file_history", None), skill_registry=getattr(seed_context, "skill_registry", None), skill_store=getattr(seed_context, "skill_store", None), sent_skill_names_by_agent={ str(agent): set(names or ()) for agent, names in ( getattr(seed_context, "sent_skill_names_by_agent", {}) or {} ).items() }, discovered_skill_names=set( getattr(seed_context, "discovered_skill_names", set()) or set() ), invoked_skills_by_agent={ str(agent): list(records or []) for agent, records in ( getattr(seed_context, "invoked_skills_by_agent", {}) or {} ).items() }, skill_listing_suppressed_once=bool( getattr(seed_context, "skill_listing_suppressed_once", False) ), active_skill_scopes=clone_skill_invocation_scopes( getattr(seed_context, "active_skill_scopes", {}) or {} ), skill_model_override=getattr(seed_context, "skill_model_override", None), skill_effort_override=getattr(seed_context, "skill_effort_override", None), dynamic_skill_path_triggers=set( getattr(seed_context, "dynamic_skill_path_triggers", set()) or set() ), sent_dynamic_skill_keys=set( getattr(seed_context, "sent_dynamic_skill_keys", set()) or set() ), path_activated_skill_names=set( getattr(seed_context, "path_activated_skill_names", set()) or set() ), skills_disabled=bool(getattr(seed_context, "skills_disabled", False)), ) @staticmethod def collect_tool_results( tool_calls: Sequence[dict[str, Any]], tool_map: dict[str, BaseTool], tool_messages: Sequence[dict[str, Any]], ) -> list[dict[str, Any]]: """Build structured tool result entries from ``run_tools()`` output.""" from openspace.tool_runtime.pipeline.execution import find_tool_by_name results: list[dict[str, Any]] = [] tool_messages_by_id: dict[str, dict[str, Any]] = {} for message in tool_messages: if message.get("role") != "tool": continue tool_call_id = str(message.get("tool_call_id") or "") if tool_call_id: tool_messages_by_id[tool_call_id] = message for tool_call in tool_calls: func = tool_call.get("function", {}) tool_name = func.get("name", "") tool_call_id = str(tool_call.get("id", "")) tool_message = tool_messages_by_id.get(tool_call_id) if tool_message is None: continue tool_obj = tool_map.get(tool_name) if tool_obj is None: tool_obj = find_tool_by_name(list(tool_map.values()), tool_name) backend = None server_name = None if tool_obj is not None: runtime_info = getattr(tool_obj, "runtime_info", None) if runtime_info is not None: backend = getattr(getattr(runtime_info, "backend", None), "value", None) server_name = getattr(runtime_info, "server_name", None) meta = tool_message.get("_meta", {}) status_raw = str(meta.get("status") or "success").strip().lower() error_statuses = {"error", "denied", "cancelled"} status = ( ToolStatus.ERROR if status_raw in error_statuses else ToolStatus.SUCCESS ) content = tool_message.get("content") error = None if status == ToolStatus.ERROR and isinstance(content, str): error = content.removeprefix("Error: ").strip() or content tool_result_kwargs: dict[str, Any] = { "status": status, "content": content, "error": error, "execution_time": meta.get("execution_time"), } metadata = meta.get("tool_result_metadata") if isinstance(metadata, dict): metadata = dict(metadata) else: metadata = {} if status_raw not in {ToolStatus.SUCCESS.value, ToolStatus.ERROR.value}: metadata["raw_status"] = status_raw if metadata: tool_result_kwargs["metadata"] = metadata tool_result = ToolResult(**tool_result_kwargs) results.append({ "tool_call": tool_call, "result": tool_result, "message": tool_message, "backend": backend, "server_name": server_name, }) return results @staticmethod def format_messages_to_text(messages: List[Dict]) -> str: """Format conversation history to readable text (for logging/debugging)""" formatted = "" for msg in messages: role = msg.get("role", "unknown").upper() content = msg.get("content", "") formatted += f"[{role}]\n{content}\n\n" return formatted