diff --git a/veadk/runtime/codex/proxy.py b/veadk/runtime/codex/proxy.py index 546c94929..6d680f737 100644 --- a/veadk/runtime/codex/proxy.py +++ b/veadk/runtime/codex/proxy.py @@ -27,7 +27,7 @@ import asyncio import json -from typing import Any, AsyncIterator, cast +from typing import Any, AsyncIterator import litellm import uvicorn @@ -95,6 +95,14 @@ async def responses(request: Request) -> Any: call_kwargs: dict[str, Any] = { key: body[key] for key in _PASSTHROUGH_KEYS if key in body } + # Codex injects its built-in tools (e.g. `web_search`) whose schema + # carries fields like `external_web_access` that non-OpenAI Responses + # backends (Ark) reject. Keep only standard `function` tools, which + # the bridged chat backend understands. + if isinstance(call_kwargs.get("tools"), list): + call_kwargs["tools"] = [ + t for t in call_kwargs["tools"] if t.get("type") == "function" + ] call_kwargs.update( model=f"openai/{model}", api_base=self.api_base, @@ -102,18 +110,23 @@ async def responses(request: Request) -> Any: custom_llm_provider="openai", drop_params=True, num_retries=0, + stream=False, ) + # Always call the backend non-streaming. litellm's chat->Responses + # bridge can only emit a single degenerate `response.completed` + # event when streaming a chat backend, which Codex's strict SSE + # parser rejects (surfaced as a generic "high demand" error). So we + # fetch the full result and, when Codex asked for a stream, + # synthesize the canonical Responses event sequence ourselves. result = await litellm.aresponses(**call_kwargs) + resp = _to_dict(result) if stream: - stream_iter = cast(AsyncIterator[Any], result).__aiter__() - first = await anext(stream_iter, None) return StreamingResponse( - _encode_sse(stream_iter, first), - media_type="text/event-stream", + _synth_sse(resp), media_type="text/event-stream" ) - return JSONResponse(_to_dict(result)) + return JSONResponse(resp) @app.exception_handler(APIError) async def _on_api_error(_request: Request, exc: APIError) -> JSONResponse: @@ -135,8 +148,15 @@ async def start(self) -> str: if self.url: return self.url + # The shim app has no startup/shutdown hooks, so disable the lifespan + # protocol; otherwise its task lingers and logs a CancelledError + # traceback when the event loop is torn down at process exit. config = uvicorn.Config( - self._app, host="127.0.0.1", port=0, log_level="warning" + self._app, + host="127.0.0.1", + port=0, + log_level="warning", + lifespan="off", ) server = uvicorn.Server(config) server.install_signal_handlers = lambda: None # type: ignore[method-assign] @@ -180,41 +200,103 @@ def _to_dict(obj: Any) -> dict[str, Any]: return dict(obj) -def _encode_chunk(chunk: Any) -> bytes: - """Encode one litellm Responses stream chunk as SSE bytes.""" - if isinstance(chunk, (bytes, bytearray)): - return bytes(chunk) - if isinstance(chunk, str): - return chunk.encode() - data = _to_dict(chunk) - event_type = data.get("type", "message") - return f"event: {event_type}\ndata: {json.dumps(data)}\n\n".encode() +def _sse(event: dict[str, Any]) -> bytes: + """Encode one Responses event dict as an SSE frame.""" + return f"event: {event['type']}\ndata: {json.dumps(event)}\n\n".encode() -async def _encode_sse( - chunks: AsyncIterator[Any], first: Any = None -) -> AsyncIterator[bytes]: - """Re-encode litellm Responses stream chunks as SSE bytes. +async def _synth_sse(resp: dict[str, Any]) -> AsyncIterator[bytes]: + """Synthesize a canonical Responses event stream from a final result. - ``first`` is the already-pulled leading chunk (or ``None``). A mid-stream - backend error is emitted as an ``error`` SSE event so the client sees a - terminal error rather than a silently truncated stream. + litellm's chat->Responses bridge cannot produce a real streamed event + sequence for a chat backend, so we expand the completed response into the + ordered events Codex expects: ``response.created`` -> per output item + (``output_item.added`` -> text/reasoning deltas -> ``output_item.done``) -> + ``response.completed``. Only ``message`` and ``reasoning`` items are + emitted (the only kinds the bridged chat backend returns); the completed + response is trimmed to match what was streamed. """ - if first is not None: - yield _encode_chunk(first) - try: - async for chunk in chunks: - yield _encode_chunk(chunk) - except APIError as exc: - status = getattr(exc, "status_code", 500) or 500 - err = { - "type": "error", - "error": { - "type": _error_type(status), - "message": getattr(exc, "message", str(exc)), - }, - } - yield f"event: error\ndata: {json.dumps(err)}\n\n".encode() + seq = 0 + + def ev(payload: dict[str, Any]) -> bytes: + nonlocal seq + payload["sequence_number"] = seq + seq += 1 + return _sse(payload) + + items = [ + it + for it in (resp.get("output") or []) + if it.get("type") in ("message", "reasoning") + ] + in_progress = {**resp, "status": "in_progress", "output": []} + yield ev({"type": "response.created", "response": in_progress}) + yield ev({"type": "response.in_progress", "response": in_progress}) + + for idx, item in enumerate(items): + item_id = item.get("id", f"item_{idx}") + stub = {**item, "status": "in_progress"} + if item.get("type") == "message": + stub = {**stub, "content": []} + else: + stub = {**stub, "summary": []} + yield ev( + {"type": "response.output_item.added", "output_index": idx, "item": stub} + ) + + if item.get("type") == "message": + for cidx, part in enumerate(item.get("content") or []): + text = part.get("text", "") + base = {"item_id": item_id, "output_index": idx, "content_index": cidx} + yield ev( + { + "type": "response.content_part.added", + **base, + "part": {"type": "output_text", "text": "", "annotations": []}, + } + ) + yield ev({"type": "response.output_text.delta", **base, "delta": text}) + yield ev({"type": "response.output_text.done", **base, "text": text}) + yield ev({"type": "response.content_part.done", **base, "part": part}) + else: # reasoning + for sidx, summary in enumerate(item.get("summary") or []): + text = summary.get("text", "") + base = {"item_id": item_id, "output_index": idx, "summary_index": sidx} + yield ev( + { + "type": "response.reasoning_summary_part.added", + **base, + "part": {"type": "summary_text", "text": ""}, + } + ) + yield ev( + { + "type": "response.reasoning_summary_text.delta", + **base, + "delta": text, + } + ) + yield ev( + { + "type": "response.reasoning_summary_text.done", + **base, + "text": text, + } + ) + yield ev( + { + "type": "response.reasoning_summary_part.done", + **base, + "part": summary, + } + ) + + yield ev( + {"type": "response.output_item.done", "output_index": idx, "item": item} + ) + + completed = {**resp, "status": "completed", "output": items} + yield ev({"type": "response.completed", "response": completed}) # Reuse one shim per (api_base, api_key) for the lifetime of the process. diff --git a/veadk/runtime/codex/translate.py b/veadk/runtime/codex/translate.py index 16902903c..5e699941e 100644 --- a/veadk/runtime/codex/translate.py +++ b/veadk/runtime/codex/translate.py @@ -68,12 +68,34 @@ def build_prompt(ctx: "InvocationContext") -> str: return "\n".join(lines) +def _reasoning_summary(result: Any) -> str: + """Join the reasoning-summary text from a result's items, if any. + + A reasoning model (e.g. DeepSeek) sometimes ends a turn with only a + ``reasoning`` item and no final ``agentMessage``, leaving + ``final_response`` empty. In that case the reasoning summary is the only + thing the model produced, so we surface it rather than print nothing. + """ + parts: list[str] = [] + for item in getattr(result, "items", None) or []: + data = item.model_dump() if hasattr(item, "model_dump") else item + if not isinstance(data, dict) or "reasoning" not in str(data.get("type", "")): + continue + for entry in data.get("summary") or []: + if isinstance(entry, str): + parts.append(entry) + elif isinstance(entry, dict) and entry.get("text"): + parts.append(str(entry["text"])) + return "\n".join(p.strip() for p in parts if p and p.strip()) + + def result_to_events(result: Any, author: str, invocation_id: str) -> list[Event]: """Convert a Codex run result into ADK events. The Codex SDK's run result exposes the assistant's final text as - ``final_response``. Richer per-item events (tool calls, reasoning) can be - added later. + ``final_response``. When a turn ends with reasoning but no final message + (``final_response`` empty), fall back to the reasoning summary so the + caller never receives a silently empty turn. Args: result (Any): The object returned by ``thread.run(...)``. @@ -82,9 +104,16 @@ def result_to_events(result: Any, author: str, invocation_id: str) -> list[Event Returns: list[google.adk.events.event.Event]: One text event, or empty if the - result carried no text. + result carried neither a final message nor reasoning. """ text = getattr(result, "final_response", None) + if not text: + summary = _reasoning_summary(result) + if summary: + text = ( + "[The model produced only reasoning, no final answer this turn]\n\n" + f"{summary}" + ) if not text: return []