diff --git a/agent/bedrock_adapter.py b/agent/bedrock_adapter.py index 0ce2c18258..8696a833b6 100644 --- a/agent/bedrock_adapter.py +++ b/agent/bedrock_adapter.py @@ -118,18 +118,22 @@ def bedrock_openai_base_url(region: str) -> str: return f"https://bedrock-mantle.{resolved}.api.aws/openai/v1" +def _mantle_url_parts(base_url: str) -> Tuple[Optional[str], str]: + """(region or None if not a Mantle host, normalized path) for a base URL.""" + parsed = urlparse(str(base_url or "")) + match = _BEDROCK_OPENAI_HOST_RE.match(parsed.hostname or "") + return (match.group(1) if match else None), (parsed.path or "").rstrip("/").lower() + + def bedrock_openai_region_from_base_url(base_url: str) -> Optional[str]: """Extract the AWS region from a Bedrock Mantle OpenAI base URL.""" - match = _BEDROCK_OPENAI_HOST_RE.match(urlparse(str(base_url or "")).hostname or "") - return match.group(1) if match else None + return _mantle_url_parts(base_url)[0] def is_bedrock_openai_base_url(base_url: str) -> bool: """True for Bedrock Mantle endpoints (bare host or /openai[/v1] path).""" - parsed = urlparse(str(base_url or "")) - if not _BEDROCK_OPENAI_HOST_RE.match(parsed.hostname or ""): - return False - return (parsed.path or "").rstrip("/").lower() in {"", "/openai", "/openai/v1"} + region, path = _mantle_url_parts(base_url) + return region is not None and path in {"", "/openai", "/openai/v1"} def resolve_bedrock_bearer_token(env: Optional[Dict[str, str]] = None) -> str: @@ -421,13 +425,12 @@ def _without_cache_points(blocks: Any) -> Optional[list]: def strip_cache_points(kwargs: Dict[str, Any], placement: str) -> Dict[str, Any]: """Copy of Converse kwargs with ``placement``'s cachePoint removed; the SAME object back when nothing was stripped (callers use identity to decide a retry cannot help).""" - if placement == "system": - cleaned = _without_cache_points(kwargs.get("system")) - return kwargs if cleaned is None else {**kwargs, "system": cleaned} - if placement == "tools": + if placement in ("system", "tools"): tool_config = kwargs.get("toolConfig") - cleaned = _without_cache_points((tool_config or {}).get("tools")) - return kwargs if cleaned is None else {**kwargs, "toolConfig": {**tool_config, "tools": cleaned}} + cleaned = _without_cache_points(kwargs.get("system") if placement == "system" else (tool_config or {}).get("tools")) + if cleaned is None: + return kwargs + return {**kwargs, "system": cleaned} if placement == "system" else {**kwargs, "toolConfig": {**tool_config, "tools": cleaned}} if placement == "messages": messages = kwargs.get("messages") if not isinstance(messages, list): @@ -479,15 +482,10 @@ def is_anthropic_bedrock_model(model_id: str) -> bool: def convert_tools_to_converse(tools: List[Dict]) -> List[Dict]: """OpenAI ``{"function": {...}}`` tool defs → Converse ``{"toolSpec": {...}}``.""" - result = [] - for t in tools or []: - fn = t.get("function", {}) - result.append({"toolSpec": { - "name": fn.get("name", ""), - "description": fn.get("description", ""), - "inputSchema": {"json": fn.get("parameters", {"type": "object", "properties": {}})}, - }}) - return result + return [{"toolSpec": { + "name": fn.get("name", ""), "description": fn.get("description", ""), + "inputSchema": {"json": fn.get("parameters", {"type": "object", "properties": {}})}, + }} for fn in (t.get("function", {}) for t in tools or [])] # Converse rejects empty OR whitespace-only text blocks, so the placeholder must be non-whitespace. @@ -513,10 +511,8 @@ def _image_block_from_data_url(url: str) -> Dict: raw_bytes = base64.b64decode(data) except Exception: raw_bytes = data.encode("utf-8") - return {"image": { - "format": media_type.split("/")[-1] if "/" in media_type else "jpeg", - "source": {"bytes": raw_bytes}, - }} + image_format = media_type.split("/")[-1] if "/" in media_type else "jpeg" + return {"image": {"format": image_format, "source": {"bytes": raw_bytes}}} def _convert_content_to_converse(content) -> List[Dict]: @@ -888,9 +884,7 @@ def build_converse_kwargs( def cache_here(placement: str) -> bool: return cache_enabled and cache_point_allowed(model, placement) - inference_config: Dict[str, Any] = {} - if max_tokens is not None: - inference_config["maxTokens"] = max_tokens + inference_config: Dict[str, Any] = {} if max_tokens is None else {"maxTokens": max_tokens} kwargs: Dict[str, Any] = {"modelId": model, "messages": converse_messages, "inferenceConfig": inference_config} if system_prompt: kwargs["system"] = system_prompt + [dict(_CACHE_POINT)] if cache_here("system") else system_prompt @@ -939,9 +933,7 @@ def call_converse( """Non-streaming Converse call → OpenAI-compatible response. Retries once without a rejected cachePoint placement; evicts the cached client on stale-connection errors.""" client = _get_bedrock_runtime_client(region) - kwargs = build_converse_kwargs( - model, messages, tools, max_tokens, temperature, top_p, stop_sequences, guardrail_config, - ) + kwargs = build_converse_kwargs(model, messages, tools, max_tokens, temperature, top_p, stop_sequences, guardrail_config) try: response = client.converse(**kwargs) except Exception as exc: @@ -1117,9 +1109,7 @@ def probe_bedrock_context_length(model_id: str, region: str) -> Optional[int]: oversized = "data " * int(tier_tokens / _WORDS_PER_TOKEN) try: client.converse( - modelId=model_id, - messages=[{"role": "user", "content": [{"text": oversized}]}], - inferenceConfig={"maxTokens": 8}, + modelId=model_id, messages=[{"role": "user", "content": [{"text": oversized}]}], inferenceConfig={"maxTokens": 8}, ) logger.debug( "Bedrock context probe for %s accepted ~%s-token prompt; " diff --git a/agent/codex_responses_adapter.py b/agent/codex_responses_adapter.py index 5914d1b142..22ecb79b8e 100644 --- a/agent/codex_responses_adapter.py +++ b/agent/codex_responses_adapter.py @@ -434,16 +434,15 @@ def _replay_tool_call_items(msg: Dict[str, Any], *, start_index: int) -> List[Di if not isinstance(tc, dict): continue fn = tc.get("function", {}) - fn_name = fn.get("name") + fn_name, arguments = fn.get("name"), fn.get("arguments", "{}") if not _nonblank(fn_name): continue call_id = _resolve_call_id( - tc.get("call_id"), tc.get("id"), fn_name, str(fn.get("arguments", "{}")), start_index + len(replayed), - canonicalize_fc=True, + tc.get("call_id"), tc.get("id"), fn_name, str(arguments), start_index + len(replayed), canonicalize_fc=True, ) replayed.append({ "type": "function_call", "call_id": _clamp_responses_call_id(call_id), - "name": _sanitize_replayed_fn_name(fn_name), "arguments": _coerce_arguments(fn.get("arguments", "{}")), + "name": _sanitize_replayed_fn_name(fn_name), "arguments": _coerce_arguments(arguments), }) return replayed @@ -584,21 +583,15 @@ def estimate_native_responses_preflight_tokens( None when native compaction is not proven eligible or conversion fails.""" if getattr(agent, "api_mode", None) != "codex_responses" or not isinstance(messages, list): return None - is_codex_backend, is_xai_responses, is_github_responses = classify_responses_route(agent) + route = classify_responses_route(agent)._asdict() from agent.native_compaction import native_compaction_context_management - if not native_compaction_context_management( - agent, is_codex_backend=is_codex_backend, is_xai_responses=is_xai_responses, - is_github_responses=is_github_responses, - ): + if not native_compaction_context_management(agent, **route): return None try: items = _chat_messages_to_responses_input( - messages, is_xai_responses=is_xai_responses, is_github_responses=is_github_responses, + messages, is_xai_responses=route["is_xai_responses"], is_github_responses=route["is_github_responses"], replay_encrypted_reasoning=bool(getattr(agent, "_codex_reasoning_replay_enabled", True)), - current_issuer_kind=_classify_responses_issuer( - is_xai_responses=is_xai_responses, is_github_responses=is_github_responses, - is_codex_backend=is_codex_backend, base_url=getattr(agent, "base_url", None), - ), + current_issuer_kind=_classify_responses_issuer(base_url=getattr(agent, "base_url", None), **route), native_compaction_eligible=True, ) except Exception: @@ -854,9 +847,7 @@ def _preflight_codex_api_kwargs( if not isinstance(tools, list): raise ValueError("Codex Responses request 'tools' must be a list when provided.") normalized_tools = [_preflight_tool(tool, idx) for idx, tool in enumerate(tools)] - if sanitize_harmony_tokens: - normalized_tools = _neutralize_harmony_structure(normalized_tools) - normalized["tools"] = normalized_tools + normalized["tools"] = _neutralize_harmony_structure(normalized_tools) if sanitize_harmony_tokens else normalized_tools if api_kwargs.get("store", False) is not False: raise ValueError("Codex Responses contract requires 'store' to be false.") for key, accept, coerce in _PREFLIGHT_OPTIONAL_FIELDS: @@ -875,16 +866,14 @@ def _preflight_codex_api_kwargs( extra_body = _optional_dict(api_kwargs, "extra_body") if extra_body: normalized["extra_body"] = dict(extra_body) - allowed_keys = set(_PREFLIGHT_ALLOWED_KEYS) - if allow_stream: - stream = api_kwargs.get("stream") - if stream is not None and stream is not True: - raise ValueError("Codex Responses 'stream' must be true when set.") - if stream is True: - normalized["stream"] = True - allowed_keys.add("stream") - elif "stream" in api_kwargs: + allowed_keys = _PREFLIGHT_ALLOWED_KEYS | ({"stream"} if allow_stream else set()) + stream = api_kwargs.get("stream") + if not allow_stream and "stream" in api_kwargs: raise ValueError("Codex Responses stream flag is only allowed in fallback streaming requests.") + if allow_stream and stream is not None and stream is not True: + raise ValueError("Codex Responses 'stream' must be true when set.") + if allow_stream and stream is True: + normalized["stream"] = True # Defense-in-depth slash-enum strip for xAI (rejects ``Qwen/Qwen3.5`` enum values); # gated on the model name because native Codex accepts slashes. is_xai_model = str(api_kwargs.get("model") or "").lower().startswith(("grok-", "x-ai/grok-")) @@ -1022,24 +1011,23 @@ class _OutputScan: reasoning_text = _extract_responses_reasoning_text(item) if reasoning_text: self.reasoning_parts.append(reasoning_text) - raw_item = _capture_reasoning_item(item, issuer_kind) - if raw_item is not None: - self.reasoning_items_raw.append(raw_item) + self._capture(_capture_reasoning_item(item, issuer_kind)) elif item_type == "compaction": # Compaction checkpoints ride the codex_reasoning_items sidecar (persistence, # replay, cross-issuer guard and kill switch for free). raw_item = _stamped_encrypted_item(item, "compaction", issuer_kind) - if raw_item is not None: - self.reasoning_items_raw.append(raw_item) - logger.info( - "Native Responses compaction item captured (%d chars encrypted).", - len(raw_item["encrypted_content"]), - ) + if self._capture(raw_item): + logger.info("Native Responses compaction item captured (%d chars encrypted).", len(raw_item["encrypted_content"])) elif item_type in {"function_call", "custom_tool_call"}: if item_type == "function_call" and item_status in _INCOMPLETE_STATUSES: continue self.tool_calls.append(_response_tool_call(item, item_type, len(self.tool_calls))) + def _capture(self, raw_item: Optional[Dict[str, Any]]) -> bool: + if raw_item is not None: + self.reasoning_items_raw.append(raw_item) + return raw_item is not None + def _message(self, item: Any, item_status: Optional[str]) -> None: normalized_phase = _lower_or_none(getattr(item, "phase", None)) is_commentary_phase = normalized_phase in {"commentary", "analysis"} diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py index 00ef3973b5..99552342ad 100644 --- a/agent/codex_runtime.py +++ b/agent/codex_runtime.py @@ -96,16 +96,15 @@ def _record_codex_app_server_usage(agent, turn) -> dict[str, Any]: agent.session_api_calls += 1 usage = getattr(turn, "token_usage_last", None) compressor = getattr(agent, "context_compressor", None) + def billing(**extra): + return dict(model=agent.model, billing_provider=agent.provider, billing_base_url=agent.base_url, api_call_count=1, **extra) if not isinstance(usage, dict) or not usage: if compressor is not None and getattr(compressor, "awaiting_real_usage_after_compression", False): # No usage cannot adjudicate the pending compaction; unlatch preflight deferral. compressor.update_from_response({}) _queue_token_counts( agent, "Codex app-server api-call persistence failed (session=%s): %s", - counts=lambda: dict( - model=agent.model, billing_provider=agent.provider, billing_base_url=agent.base_url, - billing_mode="subscription_included", api_call_count=1, - ), + counts=lambda: billing(billing_mode="subscription_included"), ) return {} from agent.usage_pricing import CanonicalUsage, estimate_usage_cost @@ -148,11 +147,8 @@ def _record_codex_app_server_usage(agent, turn) -> dict[str, Any]: cost_fields = {"estimated_cost_usd": cost_usd, "cost_status": cost_result.status, "cost_source": cost_result.source} _queue_token_counts( agent, "Codex app-server token persistence failed (session=%s, tokens=%d): %s", total_tokens, - counts=lambda: dict( - **token_counts, **cost_fields, - billing_provider=agent.provider, billing_base_url=agent.base_url, - billing_mode="subscription_included" if cost_result.status == "included" else None, - model=agent.model, api_call_count=1, + counts=lambda: billing( + **token_counts, **cost_fields, billing_mode="subscription_included" if cost_result.status == "included" else None, ), ) return {**usage_dict, "last_prompt_tokens": prompt_tokens, **cost_fields} @@ -770,8 +766,7 @@ class _CodexResponseAssembler: self.saw_terminal = True resp_obj = _event_field(event, "response") if resp_obj is not None: - self.terminal_usage = _event_field(resp_obj, "usage") - self.terminal_response_id = _event_field(resp_obj, "id") + self.terminal_usage, self.terminal_response_id = _event_field(resp_obj, "usage"), _event_field(resp_obj, "id") rstatus = _event_field(resp_obj, "status") if isinstance(rstatus, str): self.terminal_status = rstatus @@ -779,8 +774,7 @@ class _CodexResponseAssembler: self.terminal_incomplete_details = _event_field(resp_obj, "incomplete_details") elif event_type == "response.failed": self.terminal_error = _event_field(resp_obj, "error") - if event_type == "response.completed": - self.saw_response_completed = True + self.saw_response_completed = self.saw_response_completed or event_type == "response.completed" self.terminal_status = self.terminal_status or event_type.removeprefix("response.") return True @@ -1003,8 +997,7 @@ def run_codex_stream(agent, api_kwargs: dict, client: Any = None, on_first_delta _log_codex_request_failure(agent, exc, stream_opened=writer_token["value"] is not None) def _codex_stream_created(_raw_stream: Any) -> None: - # Claim the delta sink for THIS physical attempt; a newer attempt - # supersedes this token and fences late deltas out of the turn. + # Claim the delta sink for THIS attempt; a newer attempt supersedes this token. writer_token["value"] = claim_stream_writer(agent) def _accept_codex_chunk(_chunk: Any) -> bool: