diff --git a/agent/codex_responses_adapter.py b/agent/codex_responses_adapter.py index 22ecb79b8e..bcd706f136 100644 --- a/agent/codex_responses_adapter.py +++ b/agent/codex_responses_adapter.py @@ -1,8 +1,5 @@ -"""Codex Responses API adapter. - -Stateless format-conversion and normalization for the OpenAI Responses API -(OpenAI Codex, xAI, GitHub Models and other Responses-compatible endpoints). -""" +"""Codex Responses API adapter: stateless format conversion and normalization for the +OpenAI Responses API (OpenAI Codex, xAI, GitHub Models and other compatible endpoints).""" from __future__ import annotations @@ -25,9 +22,8 @@ def _classify_responses_issuer( *, is_xai_responses: bool = False, is_github_responses: bool = False, is_codex_backend: bool = False, base_url: Optional[str] = None, ) -> str: - """Stable identifier for the endpoint that mints ``reasoning.encrypted_content``. Blobs - are sealed to their issuer (HTTP 400 ``invalid_encrypted_content`` across endpoints), so - stamping lets replay drop foreign blobs after a mid-conversation model switch.""" + """Stable identifier for the endpoint that mints ``reasoning.encrypted_content``. Blobs are sealed to their + issuer (HTTP 400 ``invalid_encrypted_content``), so stamping lets replay drop foreign blobs after a model switch.""" for flag, kind in ((is_xai_responses, "xai_responses"), (is_github_responses, "github_responses"), (is_codex_backend, "codex_backend")): if flag: return kind @@ -60,8 +56,8 @@ _VALID_RESPONSES_FN_NAME_RE = re.compile(r"[a-zA-Z0-9_-]{1,64}") # Provider-executed built-in tools: declared by ``type`` alone, run server-side, # reported via the ``*_call`` output items below; preflight passes them through. _RESPONSES_BUILTIN_TOOL_TYPES = { - "web_search", "web_search_preview", "file_search", "code_interpreter", - "image_generation", "computer_use_preview", "local_shell", + "web_search", "web_search_preview", "file_search", "code_interpreter", "image_generation", "computer_use_preview", + "local_shell", } # Server-side ``*_call`` output items. xAI leaves these ``in_progress`` even when the @@ -89,6 +85,10 @@ def _lower_or_none(value: Any) -> Optional[str]: return value.strip().lower() if isinstance(value, str) else None +def _as_list(value: Any) -> list: + return value if isinstance(value, list) else [] + + def _field(obj: Any, name: str, default: Any = None) -> Any: """Read ``name`` from a dict or an attribute-style (SDK/SimpleNamespace) object.""" return obj.get(name) if isinstance(obj, dict) else getattr(obj, name, default) @@ -115,21 +115,18 @@ def _neutralize_harmony_tokens(text: str) -> str: """Keep Harmony source readable without emitting reserved wire tokens.""" if not text or "<" not in text or "|" not in text: return text - replacement = rf"<{_FULLWIDTH_PIPE}\1{_FULLWIDTH_PIPE}>" if not any(unicodedata.category(char) == "Cf" for char in text): - return _HARMONY_CONTROL_TOKEN_RE.sub(replacement, text) + return _HARMONY_CONTROL_TOKEN_RE.sub(rf"<{_FULLWIDTH_PIPE}\1{_FULLWIDTH_PIPE}>", text) # The backend strips Unicode format controls (e.g. U+200B) before its reserved-token # check, so match on the visible text and rewrite the original spans. original_positions = [i for i, char in enumerate(text) if unicodedata.category(char) != "Cf"] visible_text = "".join(text[i] for i in original_positions) - result: List[str] = [] - cursor = 0 + result, cursor = [], 0 for match in _HARMONY_CONTROL_TOKEN_RE.finditer(visible_text): start, end = original_positions[match.start()], original_positions[match.end() - 1] + 1 result += [text[cursor:start], f"<{_FULLWIDTH_PIPE}{match.group(1)}{_FULLWIDTH_PIPE}>"] cursor = end - result.append(text[cursor:]) - return "".join(result) + return "".join(result) + text[cursor:] def _neutralize_harmony_structure(value: Any) -> Any: @@ -140,12 +137,11 @@ def _neutralize_harmony_structure(value: Any) -> Any: if isinstance(value, (list, tuple)): return [_neutralize_harmony_structure(item) for item in value] if isinstance(value, dict): - for key in value: - if isinstance(key, str) and _neutralize_harmony_tokens(key) != key: - raise ValueError( - "Reserved Harmony tokens in a JSON object key cannot be " - "neutralized without changing its contract." - ) + if any(isinstance(key, str) and _neutralize_harmony_tokens(key) != key for key in value): + raise ValueError( + "Reserved Harmony tokens in a JSON object key cannot be " + "neutralized without changing its contract." + ) return {key: _neutralize_harmony_structure(item) for key, item in value.items()} return value @@ -165,94 +161,70 @@ def _iter_content_parts(content: list) -> Iterator[tuple[str, Any]]: yield "image", part -def _resolve_image_ref(part: Dict[str, Any]) -> tuple[Any, Any]: - """Return ``(url, detail)`` from either ``image_url: str`` or ``{url, detail}``.""" - image_ref = part.get("image_url") - detail = part.get("detail") - if isinstance(image_ref, dict): - return image_ref.get("url"), image_ref.get("detail", detail) - return image_ref, detail - - -def _input_image_part(url: str, detail: Any) -> Dict[str, Any]: - image_part: Dict[str, Any] = {"type": "input_image", "image_url": url} +def _input_image_part(part: Dict[str, Any], role: str = "user", *, keep_empty_url: bool) -> Optional[Dict[str, Any]]: + """Responses image part from a chat/Responses image part (``image_url`` may be a str or + ``{url, detail}``). Assistant → text placeholder (an assistant ``input_image`` 400s every + replay); user → ``input_image``, None for an empty url unless ``keep_empty_url``.""" + if role == "assistant": + return {"type": "output_text", "text": _ASSISTANT_IMAGE_PLACEHOLDER} + url, detail = part.get("image_url"), part.get("detail") + if isinstance(url, dict): + url, detail = url.get("url"), url.get("detail", detail) + if not _nonempty_str(url) and not keep_empty_url: + return None + image_part: Dict[str, Any] = {"type": "input_image", "image_url": str(url or "")} if _nonblank(detail): image_part["detail"] = detail.strip() return image_part -def _image_part_for_role(part: Dict[str, Any], role: str, *, keep_empty_url: bool) -> Optional[Dict[str, Any]]: - """Responses image part for ``role``: assistant → text placeholder (an assistant - ``input_image`` 400s every replay); user → ``input_image`` (None for an empty url unless kept).""" - if role == "assistant": - return {"type": "output_text", "text": _ASSISTANT_IMAGE_PLACEHOLDER} - url, detail = _resolve_image_ref(part) - if _nonempty_str(url): - return _input_image_part(url, detail) - return _input_image_part(str(url or ""), detail) if keep_empty_url else None - - def _chat_content_to_responses_parts(content: Any, *, role: str = "user") -> List[Dict[str, Any]]: - """Chat-style multimodal content → Responses API input parts ([] if not a list). - - Text becomes ``input_text`` (user) or ``output_text`` (assistant) — the API rejects - the wrong type per role. ``input_image`` is only legal on user messages; on assistant - messages it becomes a text marker (an assistant ``input_image`` 400s every replay).""" + """Chat-style multimodal content → Responses API input parts ([] if not a list). Text is + ``input_text`` (user) / ``output_text`` (assistant) — the API rejects the wrong type per role; + ``input_image`` is only legal on user messages (see :func:`_input_image_part`).""" text_type = _text_type_for(role) converted: List[Dict[str, Any]] = [] - for kind, payload in _iter_content_parts(content if isinstance(content, list) else []): - image_part = None if kind == "text" else _image_part_for_role(payload, role, keep_empty_url=False) + for kind, payload in _iter_content_parts(_as_list(content)): if kind == "text": converted.append({"type": text_type, "text": payload}) - elif image_part is not None: - converted.append(image_part) + elif (part := _input_image_part(payload, role, keep_empty_url=False)) is not None: + converted.append(part) return converted def _summarize_user_message_for_log(content: Any, *, sep: str = " ") -> str: - """Flatten message content to plain text: text parts joined with ``sep`` (``" "`` for - logs; ``"\\n"`` for memory providers feeding regexes), images → ``[N image(s)]`` - marker, ``""`` for None/empty, ``str(content)`` for other scalars.""" - if content is None: - return "" - if isinstance(content, str): - return content + """Flatten message content to plain text: text parts joined with ``sep`` (``" "`` for logs; ``"\\n"`` for memory + providers feeding regexes), images → ``[N image(s)]`` marker, ``""`` for None, ``str(content)`` for other scalars.""" if not isinstance(content, list): try: - return str(content) + return _str_or_empty(content) except Exception: return "" parts = list(_iter_content_parts(content)) text_bits = [payload for kind, payload in parts if kind == "text"] image_count = len(parts) - len(text_bits) - summary = sep.join(text_bits).strip() - if image_count: - note = f"[{image_count} image{'s' if image_count != 1 else ''}]" - summary = f"{note} {summary}" if summary else note - return summary + note = f"[{image_count} image{'s' if image_count != 1 else ''}]" if image_count else "" + return " ".join(bit for bit in (note, sep.join(text_bits).strip()) if bit) # --- ID helpers --------------------------------------------------------------- -def _deterministic_call_id(fn_name: str, arguments: str, index: int = 0) -> str: - """Deterministic call_id (random ids would break the prompt-cache prefix). Re-exported for run_agent/tests.""" - return deterministic_call_id(fn_name, arguments, index) +# Deterministic call_id fallback (random ids would break the prompt-cache prefix); re-exported for run_agent/tests. +_deterministic_call_id = deterministic_call_id def _clamp_responses_call_id(call_id: str) -> str: - """Keep ``call_id`` within the API's 64-char cap (the codex app-server namespaces MCP - call ids past it). The surrogate is a pure function of the original so a - ``function_call`` and its ``function_call_output`` map to the same value.""" + """Keep ``call_id`` within the API's 64-char cap (the codex app-server namespaces MCP call ids past it). The + surrogate is a pure function of the original so a ``function_call`` and its ``function_call_output`` agree.""" if len(call_id) <= _MAX_RESPONSES_ITEM_ID_LENGTH: return call_id return f"call_{hashlib.sha256(call_id.encode('utf-8', errors='replace')).hexdigest()[:32]}" def _sanitize_replayed_fn_name(name: str) -> str: - """Coerce a *replayed* ``function_call.name`` to ``^[a-zA-Z0-9_-]{1,64}$`` (an invalid - stored name 400s every later turn). Invalid runs collapse to ``_``; all-invalid → "fn". - Apply ONLY to replayed items, never live tool definitions (schema names must match the - dispatch registry); pairing with the output is by call_id.""" + """Coerce a *replayed* ``function_call.name`` to ``^[a-zA-Z0-9_-]{1,64}$`` (an invalid stored + name 400s every later turn). Invalid runs collapse to ``_``; all-invalid → "fn". Apply ONLY to + replayed items, never live tool definitions (schema names must match the dispatch registry).""" if not isinstance(name, str): return "fn" if _VALID_RESPONSES_FN_NAME_RE.fullmatch(name): @@ -301,10 +273,8 @@ def _derive_responses_function_call_id(call_id: str, response_item_id: Optional[ source = (call_id or "").strip() sanitized = re.sub(r"[^A-Za-z0-9_-]", "", source) for candidate in (source, sanitized): - if candidate.startswith("fc_"): - return candidate - if candidate.startswith("call_") and len(candidate) > len("call_"): - return f"fc_{candidate[len('call_'):]}" + if candidate.startswith("fc_") or (candidate.startswith("call_") and len(candidate) > len("call_")): + return candidate if candidate.startswith("fc_") else f"fc_{candidate[len('call_'):]}" if sanitized: return f"fc_{sanitized[:48]}" seed = source or str(response_item_id or "") or uuid.uuid4().hex @@ -315,28 +285,24 @@ def _derive_responses_function_call_id(call_id: str, response_item_id: Optional[ def _responses_tools(tools: Optional[List[Dict[str, Any]]] = None) -> Optional[List[Dict[str, Any]]]: """Convert chat-completions tool schemas to Responses function-tool schemas.""" - converted: List[Dict[str, Any]] = [] - for item in tools or []: - fn = item.get("function", {}) if isinstance(item, dict) else {} - name = fn.get("name") - if _nonblank(name): - converted.append({ - "type": "function", "name": name, "description": fn.get("description", ""), "strict": False, - "parameters": fn.get("parameters", {"type": "object", "properties": {}}), - }) + fns = [item.get("function", {}) if isinstance(item, dict) else {} for item in tools or []] + converted = [ + { + "type": "function", "name": fn["name"], "description": fn.get("description", ""), "strict": False, + "parameters": fn.get("parameters", {"type": "object", "properties": {}}), + } + for fn in fns if _nonblank(fn.get("name")) + ] return converted or None # --- Message format conversion (chat history -> Responses input) -------------- def _normalize_responses_message_status(value: Any, *, default: str = "completed") -> str: - """Normalize a replayed assistant message status, modulo case/hyphen spelling, so - incomplete Codex continuation turns are not falsely marked completed.""" - if isinstance(value, str): - status = value.strip().lower().replace("-", "_").replace(" ", "_") - if status in _RESPONSE_MESSAGE_STATUSES: - return status - return default + """Normalize a replayed assistant message status, modulo case/hyphen spelling, so incomplete Codex + continuation turns are not falsely marked completed.""" + status = value.strip().lower().replace("-", "_").replace(" ", "_") if isinstance(value, str) else None + return status if status in _RESPONSE_MESSAGE_STATUSES else default def _message_item( @@ -344,19 +310,13 @@ def _message_item( ) -> Dict[str, Any]: """Assistant ``message`` item; ``id``/``phase`` are added only when non-empty.""" item: Dict[str, Any] = {"type": "message", "role": "assistant", "status": status, "content": content} - if item_id: - item["id"] = item_id - if phase: - item["phase"] = phase + item.update({k: v for k, v in (("id", item_id), ("phase", phase)) if v}) return item -def _assistant_message_item( - raw: Dict[str, Any], content: List[Dict[str, Any]], *, is_github_responses: bool, -) -> Dict[str, Any]: - """Replayable assistant ``message`` item from a stored one. ``id`` is kept only when - short enough and never for GitHub Copilot (ids bind to a backend connection; stale → - 401); ``phase`` is preserved per OpenAI's cache guidance.""" +def _assistant_message_item(raw: Dict[str, Any], content: List[Dict[str, Any]], *, is_github_responses: bool) -> Dict[str, Any]: + """Replayable assistant ``message`` item from a stored one. ``id`` is kept only when short enough and never for + GitHub Copilot (ids bind to a backend connection; stale → 401); ``phase`` is preserved per OpenAI's cache guidance.""" item_id, phase = raw.get("id"), raw.get("phase") keep_id = not is_github_responses and _nonblank(item_id) and len(item_id.strip()) <= _MAX_RESPONSES_ITEM_ID_LENGTH return _message_item( @@ -368,24 +328,18 @@ def _assistant_message_item( def _replay_reasoning_items( msg: Dict[str, Any], *, seen_item_ids: set, current_issuer_kind: Optional[str], native_compaction_eligible: bool, ) -> List[Dict[str, Any]]: - """Replay persisted encrypted reasoning/compaction items for one assistant turn. - - Skips duplicate ids, ``compaction`` checkpoints unless THIS request carries - ``context_management`` (else a persisted checkpoint erases pre-checkpoint history on a - model that cannot decrypt it), and items stamped by another issuer (HTTP 400). - Unstamped legacy items pass. ``id`` (store=False lookups 404) and ``_issuer_kind`` are stripped.""" + """Replay persisted encrypted reasoning/compaction items for one assistant turn. Skips duplicate + ids, ``compaction`` checkpoints unless THIS request carries ``context_management`` (else a persisted + checkpoint erases pre-checkpoint history on a model that cannot decrypt it), and items stamped by + another issuer (HTTP 400); unstamped legacy items pass. ``id`` (store=False lookups 404) and + ``_issuer_kind`` are stripped.""" global _CROSS_ISSUER_WARN_EMITTED - codex_reasoning = msg.get("codex_reasoning_items") - if not isinstance(codex_reasoning, list): - return [] replayed: List[Dict[str, Any]] = [] - for ri in codex_reasoning: + for ri in _as_list(msg.get("codex_reasoning_items")): if not (isinstance(ri, dict) and ri.get("encrypted_content")): continue item_id = ri.get("id") - if item_id and item_id in seen_item_ids: - continue - if ri.get("type") == "compaction" and not native_compaction_eligible: + if (item_id and item_id in seen_item_ids) or (ri.get("type") == "compaction" and not native_compaction_eligible): continue item_issuer = ri.get("_issuer_kind") if current_issuer_kind is not None and item_issuer is not None and item_issuer != current_issuer_kind: @@ -405,18 +359,13 @@ def _replay_reasoning_items( def _replay_message_items(msg: Dict[str, Any], *, is_github_responses: bool) -> List[Dict[str, Any]]: """Replay exact assistant message items (id/phase) for prefix-cache hits.""" - codex_message_items = msg.get("codex_message_items") - if not isinstance(codex_message_items, list): - return [] replayed: List[Dict[str, Any]] = [] - for raw_item in codex_message_items: - is_assistant_message = ( - isinstance(raw_item, dict) and raw_item.get("type") == "message" - and raw_item.get("role") == "assistant" and isinstance(raw_item.get("content"), list) - ) + for raw_item in _as_list(msg.get("codex_message_items")): + if not (isinstance(raw_item, dict) and raw_item.get("type") == "message" and raw_item.get("role") == "assistant"): + continue content = [ {"type": "output_text", "text": _str_or_empty(part.get("text", ""))} - for part in (raw_item["content"] if is_assistant_message else []) + for part in _as_list(raw_item.get("content")) if isinstance(part, dict) and str(part.get("type") or "").strip() in _OUTPUT_TEXT_TYPES ] if content: @@ -426,20 +375,16 @@ def _replay_message_items(msg: Dict[str, Any], *, is_github_responses: bool) -> def _replay_tool_call_items(msg: Dict[str, Any], *, start_index: int) -> List[Dict[str, Any]]: """Convert an assistant message's ``tool_calls`` into ``function_call`` items.""" - tool_calls = msg.get("tool_calls") - if not isinstance(tool_calls, list): - return [] replayed: List[Dict[str, Any]] = [] - for tc in tool_calls: + for tc in _as_list(msg.get("tool_calls")): if not isinstance(tc, dict): continue fn = tc.get("function", {}) 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(arguments), start_index + len(replayed), canonicalize_fc=True, - ) + index = start_index + len(replayed) + call_id = _resolve_call_id(tc.get("call_id"), tc.get("id"), fn_name, str(arguments), index, 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(arguments), @@ -447,8 +392,8 @@ def _replay_tool_call_items(msg: Dict[str, Any], *, start_index: int) -> List[Di return replayed -def _tool_output_item(msg: Dict[str, Any]) -> Optional[Dict[str, Any]]: - """Convert a tool-role message to ``function_call_output`` (None if unpairable).""" +def _tool_output_items(msg: Dict[str, Any]) -> List[Dict[str, Any]]: + """Convert a tool-role message to ``[function_call_output]`` (``[]`` if unpairable).""" raw_tool_call_id = msg.get("tool_call_id") call_id, tool_response_item_id = _split_responses_tool_id(raw_tool_call_id) if not _nonblank(call_id): @@ -458,23 +403,17 @@ def _tool_output_item(msg: Dict[str, Any]) -> Optional[Dict[str, Any]]: if call_id is None and _nonblank(raw_tool_call_id): call_id = raw_tool_call_id.strip() if not _nonblank(call_id): - return None + return [] # ``output`` may be a string or an ``input_text``/``input_image`` array. tool_content = msg.get("content") - output_value: Any = ( - (_chat_content_to_responses_parts(tool_content, role="user") or "") - if isinstance(tool_content, list) else str(tool_content or "") - ) - return {"type": "function_call_output", "call_id": _clamp_responses_call_id(call_id), "output": output_value} + is_parts = isinstance(tool_content, list) + output_value: Any = (_chat_content_to_responses_parts(tool_content) or "") if is_parts else str(tool_content or "") + return [{"type": "function_call_output", "call_id": _clamp_responses_call_id(call_id), "output": output_value}] def _chat_messages_to_responses_input( - messages: List[Dict[str, Any]], - *, - is_xai_responses: bool = False, - is_github_responses: bool = False, - replay_encrypted_reasoning: bool = True, - current_issuer_kind: Optional[str] = None, + messages: List[Dict[str, Any]], *, is_xai_responses: bool = False, is_github_responses: bool = False, + replay_encrypted_reasoning: bool = True, current_issuer_kind: Optional[str] = None, native_compaction_eligible: bool = False, ) -> List[Dict[str, Any]]: """Convert internal chat-style messages to Responses input items. @@ -484,39 +423,34 @@ def _chat_messages_to_responses_input( ``AIAgent._disable_codex_reasoning_replay`` after an ``invalid_encrypted_content`` 400. ``is_github_responses``: drops ``id`` from replayed message items (Copilot 401s on stale ids). ``current_issuer_kind``: cross-issuer guard; foreign-stamped items drop, legacy items replay. - ``native_compaction_eligible``: THIS request carries ``context_management``; gates both - replaying ``compaction`` checkpoints and ``prune_pre_checkpoint_items``. Checkpoints - persist across model swaps / compression flips / resume, so without the gate one - checkpoint would delete pre-checkpoint history on a model that cannot decrypt it. - Dropping it is lossless: local history is never truncated by native compaction.""" + ``native_compaction_eligible``: THIS request carries ``context_management``; gates both replaying ``compaction`` + checkpoints and ``prune_pre_checkpoint_items``. Checkpoints persist across model swaps / compression flips / resume, + so without the gate one checkpoint would erase pre-checkpoint history on a model that cannot decrypt it (lossless: + local history is never truncated).""" items: List[Dict[str, Any]] = [] # Parallel to ``items``: source chat message per item. Pruning reads a summary # carrier's provenance from the source; the converted item may be a lossy shape. item_sources: List[Optional[Dict[str, Any]]] = [] seen_item_ids: set = set() - def emit(new_items: List[Dict[str, Any]], msg: Dict[str, Any]) -> None: items.extend(new_items) item_sources.extend([msg] * len(new_items)) - for msg in messages: if not isinstance(msg, dict): continue role = msg.get("role") if role == "tool": - tool_item = _tool_output_item(msg) - if tool_item is not None: - emit([tool_item], msg) + emit(_tool_output_items(msg), msg) continue if role not in {"user", "assistant"}: continue content = msg.get("content", "") content_parts = _chat_content_to_responses_parts(content, role=role) # [] unless a list - if isinstance(content, list): - text_type = _text_type_for(role) - content_text = "".join(p["text"] for p in content_parts if p["type"] == text_type) - else: - content_text = _str_or_empty(content) + text_type = _text_type_for(role) + content_text = ( + "".join(p["text"] for p in content_parts if p["type"] == text_type) + if isinstance(content, list) else _str_or_empty(content) + ) if role == "user": emit([{"role": role, "content": content_parts or content_text}], msg) continue @@ -533,10 +467,9 @@ def _chat_messages_to_responses_input( if fallback is not None: emit([{"role": "assistant", "content": fallback}], msg) emit(_replay_tool_call_items(msg, start_index=len(items)), msg) - # The server renders nothing placed before a compaction item, so pre-checkpoint - # history is dead weight and plaintext asks / merged summaries silently vanish. Keep - # the newest checkpoint first, retain pre-checkpoint USER and SUMMARY messages within - # a token budget, leave the tail untouched. + # The server renders nothing placed before a compaction item, so pre-checkpoint history is + # dead weight and plaintext asks / merged summaries silently vanish. Keep the newest checkpoint + # first, retain pre-checkpoint USER and SUMMARY messages within a token budget, leave the tail. if not native_compaction_eligible: return items from agent.native_compaction import prune_pre_checkpoint_items @@ -546,7 +479,6 @@ def _chat_messages_to_responses_input( class ResponsesRouteFlags(NamedTuple): """Which special Responses-API route an agent is talking to. Single owner of the codex/xai/github predicates — every site must call :func:`classify_responses_route`.""" - is_codex_backend: bool is_xai_responses: bool is_github_responses: bool @@ -560,10 +492,8 @@ def classify_responses_route(agent: Any) -> ResponsesRouteFlags: base_url = str(getattr(agent, "base_url", "") or "") hostname = str(getattr(agent, "_base_url_hostname", "") or "").lower() or base_url_hostname(base_url) lower = str(getattr(agent, "_base_url_lower", "") or base_url).lower() - def _host_is(domain: str) -> bool: return hostname == domain or hostname.endswith("." + domain) - return ResponsesRouteFlags( is_codex_backend=provider == "openai-codex" or (_host_is("chatgpt.com") and "/backend-api/codex" in lower), is_xai_responses=provider in {"xai", "xai-oauth"} or hostname == "api.x.ai", @@ -572,15 +502,10 @@ def classify_responses_route(agent: Any) -> ResponsesRouteFlags: def estimate_native_responses_preflight_tokens( - agent: Any, - messages: List[Dict[str, Any]], - *, - system_prompt: str = "", - tools: Optional[List[Dict[str, Any]]] = None, + agent: Any, messages: List[Dict[str, Any]], *, system_prompt: str = "", tools: Optional[List[Dict[str, Any]]] = None, ) -> Optional[int]: - """Estimate tokens for the checkpoint-pruned Responses payload (the full transcript - overstates a natively compacted session and fires local compression needlessly). - None when native compaction is not proven eligible or conversion fails.""" + """Estimate tokens for the checkpoint-pruned Responses payload (the full transcript overstates a natively compacted + session and fires local compression needlessly). 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 route = classify_responses_route(agent)._asdict() @@ -597,41 +522,33 @@ def estimate_native_responses_preflight_tokens( except Exception: logger.debug("native Responses preflight conversion failed; falling back to generic estimate", exc_info=True) return None - if not isinstance(items, list): - return None from agent.model_metadata import estimate_request_tokens_rough return estimate_request_tokens_rough(items, system_prompt=system_prompt or "", tools=tools) # --- Input preflight / validation -------------------------------------------- -class _PreflightCtx(NamedTuple): - sanitize_text: Callable[[str], str] - sanitize_harmony_tokens: bool - is_github_responses: bool - seen_ids: set - - -def _require_call_id(item: Dict[str, Any], idx: int, kind: str) -> str: - call_id = item.get("call_id") - if not _nonblank(call_id): - raise ValueError(f"Codex Responses input[{idx}] {kind} is missing call_id.") - return call_id.strip() +_PreflightCtx = NamedTuple("_PreflightCtx", [ + ("sanitize_text", Callable[[str], str]), ("sanitize_harmony_tokens", bool), ("is_github_responses", bool), ("seen_ids", set), +]) def _preflight_function_call(item: Dict[str, Any], idx: int, ctx: _PreflightCtx) -> Dict[str, Any]: - call_id = _require_call_id(item, idx, "function_call") - name = item.get("name") + call_id, name = item.get("call_id"), item.get("name") + if not _nonblank(call_id): + raise ValueError(f"Codex Responses input[{idx}] function_call is missing call_id.") if not _nonblank(name): raise ValueError(f"Codex Responses input[{idx}] function_call is missing name.") return { - "type": "function_call", "call_id": call_id, "name": _sanitize_replayed_fn_name(name), + "type": "function_call", "call_id": call_id.strip(), "name": _sanitize_replayed_fn_name(name), "arguments": ctx.sanitize_text(_coerce_arguments(item.get("arguments", "{}"))), } def _preflight_function_call_output(item: Dict[str, Any], idx: int, ctx: _PreflightCtx) -> Dict[str, Any]: - call_id = _require_call_id(item, idx, "function_call_output") + call_id = item.get("call_id") + if not _nonblank(call_id): + raise ValueError(f"Codex Responses input[{idx}] function_call_output is missing call_id.") output = item.get("output", "") if isinstance(output, list): # Multimodal tool result: keep recognised input_text/input_image parts, drop the rest (4xx otherwise). @@ -641,38 +558,33 @@ def _preflight_function_call_output(item: Dict[str, Any], idx: int, ctx: _Prefli if ptype == "input_text" and _nonempty_str(part.get("text")): cleaned.append({"type": "input_text", "text": ctx.sanitize_text(part["text"])}) elif ptype == "input_image" and _nonempty_str(part.get("image_url")): - cleaned.append(_input_image_part(part["image_url"], part.get("detail"))) + cleaned.append(_input_image_part(part, keep_empty_url=False)) output_value: Any = cleaned or "" else: output_value = ctx.sanitize_text(_str_or_empty(output)) - return {"type": "function_call_output", "call_id": call_id, "output": output_value} + return {"type": "function_call_output", "call_id": call_id.strip(), "output": output_value} -def _preflight_reasoning(item: Dict[str, Any], idx: int, ctx: _PreflightCtx) -> Optional[Dict[str, Any]]: +def _preflight_encrypted(item: Dict[str, Any], idx: int, ctx: _PreflightCtx) -> Optional[Dict[str, Any]]: + """``reasoning`` / ``compaction`` items: opaque, issuer-sealed; forward only API-defined fields.""" encrypted = item.get("encrypted_content") if not _nonempty_str(encrypted): return None + if item["type"] == "compaction": + return {"type": "compaction", "encrypted_content": encrypted} # ``id`` is used only for local dedup and NOT forwarded (store=False → server-side 404). item_id = item.get("id") if _nonempty_str(item_id): if item_id in ctx.seen_ids: return None ctx.seen_ids.add(item_id) - summary = item.get("summary") - if not isinstance(summary, list): - summary = [] + summary = _as_list(item.get("summary")) return { "type": "reasoning", "encrypted_content": encrypted, "summary": _neutralize_harmony_structure(summary) if ctx.sanitize_harmony_tokens else summary, } -def _preflight_compaction(item: Dict[str, Any], idx: int, ctx: _PreflightCtx) -> Optional[Dict[str, Any]]: - # Opaque, issuer-sealed checkpoint; forward only the fields the API defines. - encrypted = item.get("encrypted_content") - return {"type": "compaction", "encrypted_content": encrypted} if _nonempty_str(encrypted) else None - - def _preflight_message(item: Dict[str, Any], idx: int, ctx: _PreflightCtx) -> Dict[str, Any]: if item.get("role") != "assistant": raise ValueError(f"Codex Responses input[{idx}] message items must have role='assistant'.") @@ -694,7 +606,13 @@ def _preflight_message(item: Dict[str, Any], idx: int, ctx: _PreflightCtx) -> Di return _assistant_message_item(item, normalized_content, is_github_responses=ctx.is_github_responses) -def _preflight_role_message(item: Dict[str, Any], idx: int, role: str, ctx: _PreflightCtx) -> Dict[str, Any]: +def _preflight_role_message(item: Dict[str, Any], idx: int, ctx: _PreflightCtx) -> Dict[str, Any]: + """Untyped ``user``/``assistant`` role message — the only legal shape besides typed items.""" + role = item.get("role") + if role not in {"user", "assistant"}: + raise ValueError( + f"Codex Responses input[{idx}] has unsupported item shape (type={item.get('type')!r}, role={role!r})." + ) content = item.get("content", "") if not isinstance(content, list): return {"role": role, "content": ctx.sanitize_text(_str_or_empty(content))} @@ -706,15 +624,14 @@ def _preflight_role_message(item: Dict[str, Any], idx: int, role: str, ctx: _Pre if isinstance(part, str): if part: validated.append({"type": text_type, "text": ctx.sanitize_text(part)}) - continue - if not isinstance(part, dict): + elif not isinstance(part, dict): raise ValueError(f"Codex Responses input[{idx}].content[{part_idx}] must be an object or string.") - ptype = _part_type(part) - if ptype in _TEXT_PART_TYPES: + elif (ptype := _part_type(part)) in _TEXT_PART_TYPES: text = part.get("text", "") - validated.append({"type": text_type, "text": ctx.sanitize_text(text if isinstance(text, str) else str(text or ""))}) + text = text if isinstance(text, str) else str(text or "") + validated.append({"type": text_type, "text": ctx.sanitize_text(text)}) elif ptype in _IMAGE_PART_TYPES: - validated.append(_image_part_for_role(part, role, keep_empty_url=True)) + validated.append(_input_image_part(part, role, keep_empty_url=True)) else: raise ValueError( f"Codex Responses input[{idx}].content[{part_idx}] has unsupported type {part.get('type')!r}." @@ -723,11 +640,8 @@ def _preflight_role_message(item: Dict[str, Any], idx: int, role: str, ctx: _Pre _PREFLIGHT_ITEM_HANDLERS: Dict[str, Callable[..., Optional[Dict[str, Any]]]] = { - "function_call": _preflight_function_call, - "function_call_output": _preflight_function_call_output, - "reasoning": _preflight_reasoning, - "compaction": _preflight_compaction, - "message": _preflight_message, + "function_call": _preflight_function_call, "function_call_output": _preflight_function_call_output, + "reasoning": _preflight_encrypted, "compaction": _preflight_encrypted, "message": _preflight_message, } @@ -736,28 +650,15 @@ def _preflight_codex_input_items( ) -> List[Dict[str, Any]]: if not isinstance(raw_items, list): raise ValueError("Codex Responses input must be a list of input items.") - ctx = _PreflightCtx( - sanitize_text=_neutralize_harmony_tokens if sanitize_harmony_tokens else (lambda text: text), - sanitize_harmony_tokens=sanitize_harmony_tokens, - is_github_responses=is_github_responses, - seen_ids=set(), - ) + sanitize_text = _neutralize_harmony_tokens if sanitize_harmony_tokens else (lambda text: text) + ctx = _PreflightCtx(sanitize_text, sanitize_harmony_tokens, is_github_responses, set()) normalized: List[Dict[str, Any]] = [] for idx, item in enumerate(raw_items): if not isinstance(item, dict): raise ValueError(f"Codex Responses input[{idx}] must be an object.") item_type = item.get("type") handler = _PREFLIGHT_ITEM_HANDLERS.get(item_type) if isinstance(item_type, str) else None - if handler is not None: - normalized_item = handler(item, idx, ctx) - else: - # Untyped role messages (user/assistant) are the only other legal shape. - role = item.get("role") - if role not in {"user", "assistant"}: - raise ValueError( - f"Codex Responses input[{idx}] has unsupported item shape (type={item_type!r}, role={role!r})." - ) - normalized_item = _preflight_role_message(item, idx, role, ctx) + normalized_item = (handler or _preflight_role_message)(item, idx, ctx) if normalized_item is not None: normalized.append(normalized_item) return normalized @@ -771,12 +672,10 @@ def _preflight_tool(tool: Any, idx: int) -> Dict[str, Any]: return dict(tool) if tool_type != "function": raise ValueError(f"Codex Responses tools[{idx}] has unsupported type {tool.get('type')!r}.") - name = tool.get("name") - parameters = tool.get("parameters") - if not _nonblank(name): - raise ValueError(f"Codex Responses tools[{idx}] is missing a valid name.") - if not isinstance(parameters, dict): - raise ValueError(f"Codex Responses tools[{idx}] is missing valid parameters.") + name, parameters = tool.get("name"), tool.get("parameters") + for ok, what in ((_nonblank(name), "a valid name"), (isinstance(parameters, dict), "valid parameters")): + if not ok: + raise ValueError(f"Codex Responses tools[{idx}] is missing {what}.") return { "type": "function", "name": name.strip(), "description": _str_or_empty(tool.get("description", "")), "strict": bool(tool.get("strict", False)), "parameters": parameters, @@ -793,10 +692,10 @@ _PREFLIGHT_OPTIONAL_FIELDS: tuple[tuple[str, Callable[[Any], bool], Optional[Cal ("timeout", lambda v: isinstance(v, (int, float)) and not isinstance(v, bool) and 0 < v < float("inf"), float), ("temperature", lambda v: isinstance(v, (int, float)), float), # Cache routing/retention and tool-dispatch hints pass through as-is. - ("tool_choice", lambda v: v is not None, None), - ("parallel_tool_calls", lambda v: v is not None, None), - ("prompt_cache_key", lambda v: v is not None, None), - ("prompt_cache_retention", lambda v: v is not None, None), + *( + (key, lambda v: v is not None, None) + for key in ("tool_choice", "parallel_tool_calls", "prompt_cache_key", "prompt_cache_retention") + ), # Native compaction directive; eligibility is resolved in agent/native_compaction.py. ("context_management", lambda v: isinstance(v, list) and bool(v), None), ) @@ -815,32 +714,24 @@ def _optional_dict(api_kwargs: Dict[str, Any], key: str) -> Optional[Dict[str, A def _preflight_codex_api_kwargs( - api_kwargs: Any, - *, - allow_stream: bool = False, - is_github_responses: bool = False, + api_kwargs: Any, *, allow_stream: bool = False, is_github_responses: bool = False, sanitize_harmony_tokens: bool = False, ) -> Dict[str, Any]: if not isinstance(api_kwargs, dict): raise ValueError("Codex Responses request must be a dict.") - missing = [key for key in ("model", "instructions", "input") if key not in api_kwargs] - if missing: - raise ValueError(f"Codex Responses request missing required field(s): {', '.join(sorted(missing))}.") + if missing := sorted(key for key in ("model", "instructions", "input") if key not in api_kwargs): + raise ValueError(f"Codex Responses request missing required field(s): {', '.join(missing)}.") model = api_kwargs.get("model") if not _nonblank(model): raise ValueError("Codex Responses request 'model' must be a non-empty string.") instructions = _str_or_empty(api_kwargs.get("instructions")).strip() or DEFAULT_AGENT_IDENTITY if sanitize_harmony_tokens: instructions = _neutralize_harmony_tokens(instructions) + input_items = _preflight_codex_input_items( + api_kwargs.get("input"), is_github_responses=is_github_responses, sanitize_harmony_tokens=sanitize_harmony_tokens, + ) normalized: Dict[str, Any] = { - "model": model.strip(), - "instructions": instructions, - "input": _preflight_codex_input_items( - api_kwargs.get("input"), - is_github_responses=is_github_responses, - sanitize_harmony_tokens=sanitize_harmony_tokens, - ), - "store": False, + "model": model.strip(), "instructions": instructions, "input": input_items, "store": False, } tools = api_kwargs.get("tools") if tools is not None: @@ -854,25 +745,23 @@ def _preflight_codex_api_kwargs( value = api_kwargs.get(key) if accept(value): normalized[key] = coerce(value) if coerce else value - extra_headers = _optional_dict(api_kwargs, "extra_headers") - if extra_headers is not None: - if not all(_nonblank(key) for key in extra_headers): - raise ValueError("Codex Responses request 'extra_headers' keys must be non-empty strings.") - normalized_headers = {key.strip(): str(value) for key, value in extra_headers.items() if value is not None} - if normalized_headers: - normalized["extra_headers"] = normalized_headers + extra_headers = _optional_dict(api_kwargs, "extra_headers") or {} + if not all(_nonblank(key) for key in extra_headers): + raise ValueError("Codex Responses request 'extra_headers' keys must be non-empty strings.") + normalized_headers = {key.strip(): str(value) for key, value in extra_headers.items() if value is not None} + if normalized_headers: + normalized["extra_headers"] = normalized_headers # extra_body is verbatim: xAI carries ``prompt_cache_key`` as a body-level # field, and the SDK serializes extra_body without per-field checks. extra_body = _optional_dict(api_kwargs, "extra_body") if extra_body: normalized["extra_body"] = dict(extra_body) - 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: + if allow_stream and stream is not None: + if stream is not True: + raise ValueError("Codex Responses 'stream' must be true when set.") 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. @@ -883,8 +772,8 @@ def _preflight_codex_api_kwargs( normalized["tools"], _ = strip_slash_enum(normalized["tools"]) except Exception: pass # Best-effort — the caller-level sanitization should have handled it - unexpected = sorted(key for key in api_kwargs if key not in allowed_keys) - if unexpected: + allowed_keys = _PREFLIGHT_ALLOWED_KEYS | ({"stream"} if allow_stream else set()) + if unexpected := sorted(key for key in api_kwargs if key not in allowed_keys): raise ValueError(f"Codex Responses request has unsupported field(s): {', '.join(unexpected)}.") return normalized @@ -893,14 +782,8 @@ def _preflight_codex_api_kwargs( def _text_chunks(parts: Any, types: Optional[set] = None) -> List[str]: """Non-empty ``.text`` of each part (optionally filtered by ``.type``); [] if not a list.""" - if not isinstance(parts, list): - return [] - return [ - text for text in ( - getattr(part, "text", None) for part in parts if types is None or getattr(part, "type", None) in types - ) - if _nonempty_str(text) - ] + selected = [part for part in _as_list(parts) if types is None or getattr(part, "type", None) in types] + return [text for text in (getattr(part, "text", None) for part in selected) if _nonempty_str(text)] def _extract_responses_message_text(item: Any) -> str: @@ -921,7 +804,6 @@ def _format_responses_error(error_obj: Any, response_status: str) -> str: def field(name: str) -> str: value = _field(error_obj, name) return str(value).strip() if isinstance(value, str) or value else "" - code_str, message_str = field("code"), field("message") if code_str and message_str: return f"{code_str}: {message_str}" @@ -930,10 +812,6 @@ def _format_responses_error(error_obj: Any, response_status: str) -> str: # --- Full response normalization ---------------------------------------------- -def _synthetic_message(content: List[Any]) -> SimpleNamespace: - return SimpleNamespace(type="message", role="assistant", status="completed", content=content) - - def _response_tool_call(item: Any, item_type: str, index: int) -> SimpleNamespace: """Build a chat-style tool_call from a ``function_call``/``custom_tool_call`` item.""" fn_name = getattr(item, "name", "") or "" @@ -941,9 +819,7 @@ def _response_tool_call(item: Any, item_type: str, index: int) -> SimpleNamespac if not isinstance(arguments, str): arguments = json.dumps(arguments, ensure_ascii=False) raw_item_id = getattr(item, "id", None) - call_id = _resolve_call_id( - getattr(item, "call_id", None), raw_item_id, fn_name, arguments, index, canonicalize_fc=False, - ) + call_id = _resolve_call_id(getattr(item, "call_id", None), raw_item_id, fn_name, arguments, index, canonicalize_fc=False) fc_id = _derive_responses_function_call_id(call_id, raw_item_id if isinstance(raw_item_id, str) else None) return SimpleNamespace( id=call_id, call_id=call_id, response_item_id=fc_id, type="function", @@ -951,34 +827,27 @@ def _response_tool_call(item: Any, item_type: str, index: int) -> SimpleNamespac ) -def _stamped_encrypted_item(item: Any, item_type: str, issuer_kind: Optional[str]) -> Optional[Dict[str, Any]]: - """``{type, encrypted_content[, _issuer_kind]}`` for replay, or None without a blob.""" +def _capture_encrypted_item(item: Any, item_type: str, issuer_kind: Optional[str]) -> Optional[Dict[str, Any]]: + """``{type, encrypted_content[, _issuer_kind]}`` for replay, or None without a blob. Reasoning + items also carry ``id`` + ``summary`` (required by the API on replay); transient ``rs_tmp_`` skip.""" encrypted = getattr(item, "encrypted_content", None) if not _nonempty_str(encrypted): return None raw_item: Dict[str, Any] = {"type": item_type, "encrypted_content": encrypted} if issuer_kind: raw_item["_issuer_kind"] = issuer_kind - return raw_item - - -def _capture_reasoning_item(item: Any, issuer_kind: Optional[str]) -> Optional[Dict[str, Any]]: - """Capture a reasoning item (blob + summary) for replay; transient ``rs_tmp_`` items are skipped.""" - raw_item = _stamped_encrypted_item(item, "reasoning", issuer_kind) - if raw_item is None: - return None + if item_type != "reasoning": + return raw_item item_id = getattr(item, "id", None) if isinstance(item_id, str) and item_id.startswith("rs_tmp_"): logger.debug("Skipping transient Codex reasoning item during normalization: %s", item_id) return None if _nonempty_str(item_id): raw_item["id"] = item_id - # Summary is required by the API when replaying reasoning items. summary = getattr(item, "summary", None) if isinstance(summary, list): - raw_item["summary"] = [ - {"type": "summary_text", "text": text} for text in (getattr(part, "text", None) for part in summary) if isinstance(text, str) - ] + texts = (getattr(part, "text", None) for part in summary) + raw_item["summary"] = [{"type": "summary_text", "text": text} for text in texts if isinstance(text, str)] return raw_item @@ -986,16 +855,11 @@ class _OutputScan: """Accumulated view of one Responses ``output`` list (phase 1 of normalization).""" def __init__(self, response_status: Optional[str]) -> None: - self.content_parts: List[str] = [] - self.reasoning_parts: List[str] = [] - self.reasoning_items_raw: List[Dict[str, Any]] = [] - self.message_items_raw: List[Dict[str, Any]] = [] - self.tool_calls: List[Any] = [] + self.content_parts, self.reasoning_parts, self.tool_calls = [], [], [] + self.reasoning_items_raw, self.message_items_raw = [], [] self.has_incomplete_items = response_status in _INCOMPLETE_STATUSES self.saw_streaming_or_item_incomplete = response_status in {"queued", "in_progress"} - self.saw_commentary_phase = False - self.saw_final_answer_phase = False - self.saw_reasoning_item = False + self.saw_commentary_phase = self.saw_final_answer_phase = self.saw_reasoning_item = False def scan(self, output: List[Any], issuer_kind: Optional[str]) -> None: for item in output: @@ -1006,28 +870,24 @@ class _OutputScan: self.saw_streaming_or_item_incomplete = True if item_type == "message": self._message(item, item_status) - elif item_type == "reasoning": - self.saw_reasoning_item = True - reasoning_text = _extract_responses_reasoning_text(item) - if reasoning_text: - self.reasoning_parts.append(reasoning_text) - self._capture(_capture_reasoning_item(item, issuer_kind)) - elif item_type == "compaction": + elif item_type in {"reasoning", "compaction"}: + if item_type == "reasoning": + self.saw_reasoning_item = True + reasoning_text = _extract_responses_reasoning_text(item) + if reasoning_text: + self.reasoning_parts.append(reasoning_text) # 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 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 + raw_item = _capture_encrypted_item(item, item_type, issuer_kind) + if raw_item is not None: + self.reasoning_items_raw.append(raw_item) + if item_type == "compaction": + logger.info( + "Native Responses compaction item captured (%d chars encrypted).", len(raw_item["encrypted_content"]), + ) + elif item_type == "custom_tool_call" or (item_type == "function_call" and item_status not in _INCOMPLETE_STATUSES): 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"} @@ -1041,8 +901,7 @@ class _OutputScan: (self.reasoning_parts if is_commentary_phase else self.content_parts).append(message_text) item_id = getattr(item, "id", None) self.message_items_raw.append(_message_item( - [{"type": "output_text", "text": message_text}], - status=_normalize_responses_message_status(item_status), + [{"type": "output_text", "text": message_text}], status=_normalize_responses_message_status(item_status), item_id=item_id if isinstance(item_id, str) else None, phase=normalized_phase, )) @@ -1051,26 +910,25 @@ def _normalize_codex_response(response: Any, *, issuer_kind: Optional[str] = Non """Normalize a Responses API object to ``(assistant_message, finish_reason)``. ``issuer_kind`` is stamped onto captured reasoning items for cross-issuer replay drops.""" response_status = _lower_or_none(getattr(response, "status", None)) - incomplete_reason = _field(getattr(response, "incomplete_details", None), "reason", "") - response_incomplete_content_filter = ( - response_status == "incomplete" and str(incomplete_reason or "").strip().lower() == "content_filter" - ) + incomplete_reason = str(_field(getattr(response, "incomplete_details", None), "reason", "") or "").strip().lower() + response_incomplete_content_filter = response_status == "incomplete" and incomplete_reason == "content_filter" output = getattr(response, "output", None) if not isinstance(output, list) or not output: # Codex can deliver the whole answer via stream events with an empty output. out_text = getattr(response, "output_text", None) - if isinstance(out_text, str) and out_text.strip(): - logger.debug( - "Codex response has empty output but output_text is present (%d chars); synthesizing output item.", - len(out_text.strip()), - ) - output = [_synthetic_message([SimpleNamespace(type="output_text", text=out_text.strip())])] + out_text = out_text.strip() if isinstance(out_text, str) else "" + if out_text: + msg = "Codex response has empty output but output_text is present (%d chars); synthesizing output item." + logger.debug(msg, len(out_text)) + content: List[Any] = [SimpleNamespace(type="output_text", text=out_text)] elif response_incomplete_content_filter: # Provider safety block, not a partial answer: finish content_filter, not incomplete. - output = [_synthetic_message([])] + content = [] else: raise RuntimeError("Responses API returned no output items") - response.output = output + response.output = output = [ + SimpleNamespace(type="message", role="assistant", status="completed", content=content), + ] if response_status in {"failed", "cancelled"}: raise RuntimeError(_format_responses_error(getattr(response, "error", None), response_status)) scan = _OutputScan(response_status) @@ -1080,9 +938,8 @@ def _normalize_codex_response(response: Any, *, issuer_kind: Optional[str] = Non if not final_text and (scan.saw_final_answer_phase or not scan.saw_commentary_phase): out_text = getattr(response, "output_text", "") final_text = out_text.strip() if isinstance(out_text, str) else final_text - # Tool-call leak recovery: gpt-5.x sometimes emits the intended ``function_call`` as - # plain Harmony text (``to=functions.foo {json}``). Treat as incomplete so the - # continuation re-elicits a real call; clear the text so the garbage is not surfaced. + # Tool-call leak recovery: gpt-5.x sometimes emits the intended ``function_call`` as plain Harmony text + # (``to=functions.foo {json}``). Treat as incomplete so the continuation re-elicits a real call; clear the garbage. leaked_tool_call_text = bool(final_text and not tool_calls and _TOOL_CALL_LEAK_PATTERN.search(final_text)) if leaked_tool_call_text: logger.warning( @@ -1091,30 +948,31 @@ def _normalize_codex_response(response: Any, *, issuer_kind: Optional[str] = Non "Leaked snippet: %r", final_text[:300], ) final_text = "" - # xAI grok-4.x sometimes puts the final answer inside the reasoning item after a - # ```` delimiter; without salvage the reasoning-only rule marks the turn - # incomplete and every continuation is byte-identical. Promote the tail to content. + # xAI grok-4.x sometimes puts the final answer inside the reasoning item after a ```` delimiter; without + # salvage the reasoning-only rule marks the turn incomplete and every continuation is byte-identical. Promote the tail. if issuer_kind == "xai_responses" and not final_text and not tool_calls and reasoning_parts: joined_reasoning = "\n\n".join(reasoning_parts) marker = joined_reasoning.rfind("") - if marker != -1: - salvaged = joined_reasoning[marker + len(""):].split("", 1)[0].strip() - if salvaged: - logger.warning( - "xAI response delivered its final answer inside the reasoning channel " - "( delimiter); promoting %d chars to assistant content.", len(salvaged), - ) - final_text = salvaged - reasoning_prefix = joined_reasoning[:marker].strip() - reasoning_parts = [reasoning_prefix] if reasoning_prefix else [] + salvaged = joined_reasoning[marker + len(""):].split("", 1)[0].strip() if marker != -1 else "" + if salvaged: + logger.warning( + "xAI response delivered its final answer inside the reasoning channel " + "( delimiter); promoting %d chars to assistant content.", len(salvaged), + ) + final_text = salvaged + reasoning_prefix = joined_reasoning[:marker].strip() + reasoning_parts = [reasoning_prefix] if reasoning_prefix else [] assistant_message = SimpleNamespace( - content=final_text, - tool_calls=tool_calls, + content=final_text, tool_calls=tool_calls, reasoning="\n\n".join(reasoning_parts).strip() if reasoning_parts else None, - reasoning_content=None, - reasoning_details=None, - codex_reasoning_items=scan.reasoning_items_raw or None, - codex_message_items=scan.message_items_raw or None, + reasoning_content=None, reasoning_details=None, + codex_reasoning_items=scan.reasoning_items_raw or None, codex_message_items=scan.message_items_raw or None, + ) + # Reasoning-only: for Codex/xAI/GitHub, status=completed means "still thinking" → incomplete so the continuation + # retries. Other backends trust response.status — forcing incomplete there stalls for minutes on a final state. + reasoning_only = (scan.reasoning_items_raw or reasoning_parts or scan.saw_reasoning_item) and not final_text + trusted_final = ( + response_status == "completed" and issuer_kind not in ("codex_backend", "xai_responses", "github_responses") ) if tool_calls: finish_reason = "tool_calls" @@ -1124,16 +982,9 @@ def _normalize_codex_response(response: Any, *, issuer_kind: Optional[str] = Non leaked_tool_call_text or scan.saw_streaming_or_item_incomplete or ((scan.has_incomplete_items or scan.saw_commentary_phase) and not scan.saw_final_answer_phase) + or (reasoning_only and not trusted_final) ): finish_reason = "incomplete" - elif (scan.reasoning_items_raw or reasoning_parts or scan.saw_reasoning_item) and not final_text: - # Reasoning-only: for Codex/xAI/GitHub, status=completed means "still thinking" → - # incomplete so the continuation retries. Other backends trust response.status — - # forcing incomplete there stalls for minutes on a legitimately final state. - trusted_final = ( - response_status == "completed" and issuer_kind not in ("codex_backend", "xai_responses", "github_responses") - ) - finish_reason = "stop" if trusted_final else "incomplete" else: finish_reason = "stop" return assistant_message, finish_reason