diff --git a/.gitignore b/.gitignore index a16646df..948884e6 100644 --- a/.gitignore +++ b/.gitignore @@ -202,3 +202,4 @@ tests/e2e/node_modules/ tests/e2e/playwright-report/ tests/e2e/test-results/ tests/e2e/.e2e-last-run.json +application/core/models/foundry_test_internal.yaml diff --git a/application/agents/base.py b/application/agents/base.py index 243eb3f6..061b726c 100644 --- a/application/agents/base.py +++ b/application/agents/base.py @@ -145,24 +145,86 @@ class BaseAgent(ABC): yield from self._emit_responses_metadata() def _emit_responses_metadata(self) -> Generator[Dict, None, None]: - """Surface the latest Responses API id so the route can persist it in - message metadata for previous_response_id chaining across turns.""" - if not settings.OPENAI_RESPONSES_STORE: + """Surface Responses continuity and usage for durable next turns.""" + uses_responses = getattr(self.llm, "_uses_responses_api", None) + if callable(uses_responses) and not uses_responses(): return response_id = getattr(self.llm, "_last_response_id", None) - if response_id: - yield {"metadata": {"response_id": response_id}} + chain_key_factory = getattr(self.llm, "responses_chain_key", None) + chain_key = chain_key_factory() if callable(chain_key_factory) else None + exporter = getattr(self.llm, "export_responses_state", None) + state = exporter() if callable(exporter) else None + stored_metadata = ( + { + "response_id": response_id, + "response_chain_key": chain_key, + } + if settings.OPENAI_RESPONSES_STORE + else {} + ) + metadata = { + **stored_metadata, + "responses_state": state, + "usage": getattr(self.llm, "_last_usage", None), + } + metadata = {key: value for key, value in metadata.items() if value is not None} + if metadata: + yield {"metadata": metadata} def _previous_response_id(self) -> Optional[str]: - """Most recent stored Responses API id from chat history, if any.""" - for turn in reversed(self.chat_history or []): - if not isinstance(turn, dict): - continue - meta = turn.get("metadata") - if isinstance(meta, dict) and meta.get("response_id"): - return meta["response_id"] + """Return the immediately preceding compatible Responses API id.""" + if not self.chat_history: + return None + turn = self.chat_history[-1] + if not isinstance(turn, dict): + return None + meta = turn.get("metadata") + if not isinstance(meta, dict): + return None + chain_key_factory = getattr(self.llm, "responses_chain_key", None) + current_chain_key = ( + chain_key_factory() if callable(chain_key_factory) else None + ) + if ( + current_chain_key + and meta.get("response_chain_key") == current_chain_key + and meta.get("response_id") + ): + return meta["response_id"] return None + def _previous_responses_state(self) -> Optional[Dict[str, Any]]: + """Return continuity state from the immediately preceding turn.""" + if not self.chat_history or not isinstance(self.chat_history[-1], dict): + return None + metadata = self.chat_history[-1].get("metadata") + if not isinstance(metadata, dict): + return None + state = metadata.get("responses_state") + return state if isinstance(state, dict) else None + + def _compatible_responses_state( + self, metadata: Any + ) -> Optional[Dict[str, Any]]: + """Return Responses state only for the active Responses target.""" + uses_responses = getattr(self.llm, "_uses_responses_api", None) + if not callable(uses_responses) or not uses_responses(): + return None + if not isinstance(metadata, dict): + return None + state = metadata.get("responses_state") + chain_key_factory = getattr(self.llm, "responses_chain_key", None) + current_chain_key = ( + chain_key_factory() if callable(chain_key_factory) else None + ) + if ( + not isinstance(state, dict) + or not current_chain_key + or state.get("chain_key") != current_chain_key + ): + return None + return state + @abstractmethod def _gen_inner( self, query: str, log_context: LogContext @@ -308,7 +370,7 @@ class BaseAgent(ABC): } # Resume the LLM loop with the updated messages - llm_response = self._llm_gen(messages) + llm_response = self._llm_gen(messages, preserve_responses_state=True) yield from self._handle_response( llm_response, tools_dict, messages, None ) @@ -475,8 +537,87 @@ class BaseAgent(ABC): messages = [{"role": "system", "content": system_prompt}] for i in working_history: - if "prompt" in i and "response" in i: + has_completed_turn = "prompt" in i and "response" in i + if has_completed_turn: messages.append({"role": "user", "content": i["prompt"]}) + state = self._compatible_responses_state(i.get("metadata")) + historical_tool_calls = i.get("tool_calls") or [] + if historical_tool_calls: + tool_message: Dict[str, Any] = { + "role": "assistant", + "content": None, + "tool_calls": [], + } + call_reasoning: List[Dict[str, Any]] = [] + seen_reasoning_ids = set() + used_replay_call_ids: set[str] = set() + call_id_occurrences: Dict[str, int] = {} + for tool_call in historical_tool_calls: + # Persistence flattens all tool rounds in a turn. Some + # providers reuse deterministic call IDs in later rounds, + # so retain the first ID and synthesize stable replay-only + # IDs for collisions without dropping any call or result. + source_call_id = str( + tool_call.get("call_id") or uuid.uuid4() + ) + occurrence = call_id_occurrences.get(source_call_id, 0) + call_id_occurrences[source_call_id] = occurrence + 1 + call_id = source_call_id + while call_id in used_replay_call_ids: + occurrence += 1 + call_id = "replay_" + str(uuid.uuid5( + uuid.NAMESPACE_OID, + f"{source_call_id}:{occurrence}", + )) + used_replay_call_ids.add(call_id) + args = tool_call.get("arguments") + args_str = ( + json.dumps(args) + if isinstance(args, dict) + else (args or "{}") + ) + tool_message["tool_calls"].append({ + "id": call_id, + "type": "function", + "function": { + "name": tool_call.get("action_name", ""), + "arguments": args_str, + }, + }) + if state: + for reasoning_item in ( + state.get("reasoning_for_calls", {}).get( + source_call_id, [] + ) + ): + reasoning_id = ( + reasoning_item.get("id") + if isinstance(reasoning_item, dict) + else None + ) + if reasoning_id and reasoning_id in seen_reasoning_ids: + continue + if reasoning_id: + seen_reasoning_ids.add(reasoning_id) + call_reasoning.append(reasoning_item) + if call_reasoning: + tool_message["responses_reasoning_items"] = call_reasoning + messages.append(tool_message) + for tool_call, emitted_call in zip( + historical_tool_calls, tool_message["tool_calls"] + ): + result = tool_call.get("result") + result_str = ( + json.dumps(result) + if not isinstance(result, str) + else (result or "") + ) + messages.append({ + "role": "tool", + "tool_call_id": emitted_call["id"], + "content": result_str, + }) + if has_completed_turn: asst_msg: Dict[str, Any] = { "role": "assistant", "content": i["response"], @@ -487,39 +628,9 @@ class BaseAgent(ABC): # request. Other OpenAI-compatible APIs ignore the field. if i.get("thought"): asst_msg["reasoning_content"] = i["thought"] + if isinstance(state, dict) and state.get("reasoning_items"): + asst_msg["responses_reasoning_items"] = state["reasoning_items"] messages.append(asst_msg) - if "tool_calls" in i: - for tool_call in i["tool_calls"]: - call_id = tool_call.get("call_id") or str(uuid.uuid4()) - args = tool_call.get("arguments") - args_str = ( - json.dumps(args) - if isinstance(args, dict) - else (args or "{}") - ) - messages.append({ - "role": "assistant", - "content": None, - "tool_calls": [{ - "id": call_id, - "type": "function", - "function": { - "name": tool_call.get("action_name", ""), - "arguments": args_str, - }, - }], - }) - result = tool_call.get("result") - result_str = ( - json.dumps(result) - if not isinstance(result, str) - else (result or "") - ) - messages.append({ - "role": "tool", - "tool_call_id": call_id, - "content": result_str, - }) # When the request was multimodal, send the full content array (text + # image_url parts) so images reach the model; the text-only `query` above # is used only for token budgeting / retrieval. @@ -577,9 +688,19 @@ class BaseAgent(ABC): # ---- LLM generation ---- - def _llm_gen(self, messages: List[Dict], log_context: Optional[LogContext] = None): + def _llm_gen( + self, + messages: List[Dict], + log_context: Optional[LogContext] = None, + preserve_responses_state: bool = False, + ): self._validate_context_size(messages) + if not preserve_responses_state: + starter = getattr(self.llm, "start_responses_turn", None) + if callable(starter): + starter() + # Use the upstream id resolved by LLMCreator (see __init__). # Built-in models: same as self.model_id. BYOM: the user's # typed model name, not the internal UUID. diff --git a/application/api/answer/routes/base.py b/application/api/answer/routes/base.py index 81631f6b..c4817462 100644 --- a/application/api/answer/routes/base.py +++ b/application/api/answer/routes/base.py @@ -704,6 +704,20 @@ class BaseAnswerResource: "reasoning_content": continuation.get( "reasoning_content", "" ), + # OpenAI Responses continuity. This contains + # only upstream ids and encrypted reasoning + # blobs, never plaintext chain-of-thought. + "responses_state": ( + agent.llm.export_responses_state() + if callable( + getattr( + agent.llm, + "export_responses_state", + None, + ) + ) + else None + ), }, client_tools=getattr( agent.tool_executor, "client_tools", None @@ -782,29 +796,34 @@ class BaseAnswerResource: if isNoneDoc: for doc in source_log_docs: doc["source"] = "None" - # Model-owner scope so title-gen uses owner's BYOM key. - provider = ( - get_provider_from_model_id( - model_id, - user_id=model_user_id - or (decoded_token.get("sub") if decoded_token else None), + # Hidden API conversations keep the deterministic fallback title. + # Do not put an extra (potentially high-reasoning) LLM request on + # their response critical path. + llm = None + if visibility == "listed": + title_model_id = settings.TITLE_MODEL_ID or model_id + provider = ( + get_provider_from_model_id( + title_model_id, + user_id=model_user_id + or (decoded_token.get("sub") if decoded_token else None), + ) + if title_model_id + else settings.LLM_PROVIDER ) - if model_id - else settings.LLM_PROVIDER - ) - system_api_key = get_api_key_for_provider(provider or settings.LLM_PROVIDER) - - llm = LLMCreator.create_llm( - provider or settings.LLM_PROVIDER, - api_key=system_api_key, - user_api_key=user_api_key, - decoded_token=decoded_token, - model_id=model_id, - agent_id=agent_id, - model_user_id=model_user_id, - ) - # Title-gen only; agent stream tokens live on ``agent.llm``. - llm._token_usage_source = "title" + system_api_key = get_api_key_for_provider( + provider or settings.LLM_PROVIDER + ) + llm = LLMCreator.create_llm( + provider or settings.LLM_PROVIDER, + api_key=system_api_key, + user_api_key=user_api_key, + decoded_token=decoded_token, + model_id=title_model_id, + agent_id=agent_id, + model_user_id=model_user_id, + ) + llm._token_usage_source = "title" if should_persist: if reserved_message_id is not None: @@ -825,7 +844,8 @@ class BaseAnswerResource: "fallback_name": ( question[:50] if question else "New Conversation" ), - }, + } if llm is not None else None, + async_title_generation=llm is not None, ) else: conversation_id = self.conversation_service.save_conversation( @@ -957,34 +977,35 @@ class BaseAnswerResource: if isNoneDoc: for doc in source_log_docs: doc["source"] = "None" - # Resolve under model-owner scope so shared-agent - # title-gen uses owner BYOM, not deployment default. - provider = ( - get_provider_from_model_id( - model_id, - user_id=model_user_id - or ( - decoded_token.get("sub") - if decoded_token - else None - ), + llm = None + if visibility == "listed": + title_model_id = settings.TITLE_MODEL_ID or model_id + provider = ( + get_provider_from_model_id( + title_model_id, + user_id=model_user_id + or ( + decoded_token.get("sub") + if decoded_token + else None + ), + ) + if title_model_id + else settings.LLM_PROVIDER ) - if model_id - else settings.LLM_PROVIDER - ) - sys_api_key = get_api_key_for_provider( - provider or settings.LLM_PROVIDER - ) - llm = LLMCreator.create_llm( - provider or settings.LLM_PROVIDER, - api_key=sys_api_key, - user_api_key=user_api_key, - decoded_token=decoded_token, - model_id=model_id, - agent_id=agent_id, - model_user_id=model_user_id, - ) - llm._token_usage_source = "title" + sys_api_key = get_api_key_for_provider( + provider or settings.LLM_PROVIDER + ) + llm = LLMCreator.create_llm( + provider or settings.LLM_PROVIDER, + api_key=sys_api_key, + user_api_key=user_api_key, + decoded_token=decoded_token, + model_id=title_model_id, + agent_id=agent_id, + model_user_id=model_user_id, + ) + llm._token_usage_source = "title" if reserved_message_id is not None: outcome = self.conversation_service.finalize_message( reserved_message_id, @@ -1003,7 +1024,8 @@ class BaseAnswerResource: "fallback_name": ( question[:50] if question else "New Conversation" ), - }, + } if llm is not None else None, + async_title_generation=llm is not None, ) # ``ALREADY_COMPLETE`` means the normal-path # finalize at line 632 won the race: the DB row diff --git a/application/api/answer/services/compression/types.py b/application/api/answer/services/compression/types.py index 7bdba590..cfd77618 100644 --- a/application/api/answer/services/compression/types.py +++ b/application/api/answer/services/compression/types.py @@ -87,5 +87,7 @@ class CompressionResult: } if q.get("thought"): entry["thought"] = q["thought"] + if q.get("tool_calls"): + entry["tool_calls"] = q["tool_calls"] out.append(entry) return out diff --git a/application/api/answer/services/continuation_service.py b/application/api/answer/services/continuation_service.py index ff7fe03a..20ce8d93 100644 --- a/application/api/answer/services/continuation_service.py +++ b/application/api/answer/services/continuation_service.py @@ -23,7 +23,16 @@ PENDING_STATE_TTL_SECONDS = 30 * 60 # 30 minutes # Re-export so the existing tests at tests/api/answer/services/test_continuation_service_pg.py # can keep importing ``_make_serializable`` from here. -__all__ = ["_make_serializable", "ContinuationService", "PENDING_STATE_TTL_SECONDS"] +__all__ = [ + "_make_serializable", + "ContinuationService", + "PENDING_STATE_TTL_SECONDS", + "ResumeInProgressError", +] + + +class ResumeInProgressError(ValueError): + """Raised when another request already owns a continuation claim.""" class ContinuationService: @@ -120,6 +129,29 @@ class ContinuationService: return None return doc + def claim_state( + self, conversation_id: str, user: str + ) -> Optional[Dict[str, Any]]: + """Atomically claim live state or classify an active duplicate.""" + with db_session() as conn: + conv = ConversationsRepository(conn).get_by_legacy_id(conversation_id) + if conv is not None: + pg_conv_id = conv["id"] + elif looks_like_uuid(conversation_id): + pg_conv_id = conversation_id + else: + return None + repo = PendingToolStateRepository(conn) + claimed = repo.claim_state(pg_conv_id, user) + if claimed: + return claimed + existing = repo.load_state_any(pg_conv_id, user) + if existing and existing.get("status") == "resuming": + raise ResumeInProgressError( + "Resume already in progress for this conversation." + ) + return None + def delete_state(self, conversation_id: str, user: str) -> bool: """Delete pending state after successful resumption. diff --git a/application/api/answer/services/conversation_service.py b/application/api/answer/services/conversation_service.py index 24da1b0f..1ca64955 100644 --- a/application/api/answer/services/conversation_service.py +++ b/application/api/answer/services/conversation_service.py @@ -6,6 +6,7 @@ than held for the duration of a stream. """ import logging +import threading import uuid from datetime import datetime, timezone from typing import Any, Dict, List, Optional @@ -151,16 +152,13 @@ class ConversationService: # records). The LLM's own ``model_id`` is the upstream name # LLMCreator resolved at construction time — that's what # the provider's API expects. Built-ins are unaffected. - completion = llm.gen( - model=getattr(llm, "model_id", None) or model_id, - messages=messages_summary, - # Reasoning-capable default models spend the whole budget inside - # reasoning_content before emitting any title, so 500 came back - # empty (finish_reason=length). Give enough room to finish - # thinking and still produce the 3-word title; non-reasoning - # models stop far short of this cap. - max_tokens=2000, - ) + completion = None + if llm is not None: + completion = llm.gen( + model=getattr(llm, "model_id", None) or model_id, + messages=messages_summary, + max_tokens=2000, + ) if not completion or not completion.strip(): completion = question[:50] if question else "New Conversation" @@ -317,6 +315,7 @@ class ConversationService: status: str = "complete", error: Optional[BaseException] = None, title_inputs: Optional[Dict[str, Any]] = None, + async_title_generation: bool = False, ) -> MessageUpdateOutcome: """Commit the response and tool_call confirms in one transaction. @@ -369,16 +368,30 @@ class ConversationService: # Outside the txn — title-gen is a multi-second LLM round trip. if title_inputs and status == "complete": - try: - with db_session() as conn: - self._maybe_generate_title(conn, message_id, title_inputs) - except Exception as e: - logger.error( - f"finalize_message title generation failed: {e}", - exc_info=True, - ) + if async_title_generation: + threading.Thread( + target=self._generate_title_safely, + args=(message_id, title_inputs), + name="docsgpt-title-generation", + daemon=True, + ).start() + else: + self._generate_title_safely(message_id, title_inputs) return MessageUpdateOutcome.UPDATED + def _generate_title_safely( + self, message_id: str, title_inputs: Dict[str, Any] + ) -> None: + """Generate a title in its own DB transaction without surfacing errors.""" + try: + with db_session() as conn: + self._maybe_generate_title(conn, message_id, title_inputs) + except Exception as e: + logger.error( + f"finalize_message title generation failed: {e}", + exc_info=True, + ) + def _maybe_generate_title( self, conn, diff --git a/application/api/answer/services/stream_processor.py b/application/api/answer/services/stream_processor.py index cfd535c1..e06a6dd5 100644 --- a/application/api/answer/services/stream_processor.py +++ b/application/api/answer/services/stream_processor.py @@ -304,6 +304,11 @@ class StreamProcessor: if "metadata" in query else {} ), + **( + {"tool_calls": query["tool_calls"]} + if query.get("tool_calls") + else {} + ), } for query in conversation.get("queries", []) ] @@ -341,6 +346,11 @@ class StreamProcessor: else {} ), **({"metadata": query["metadata"]} if "metadata" in query else {}), + **( + {"tool_calls": query["tool_calls"]} + if query.get("tool_calls") + else {} + ), } for query in conversation.get("queries", []) ] @@ -381,6 +391,11 @@ class StreamProcessor: else {} ), **({"metadata": query["metadata"]} if "metadata" in query else {}), + **( + {"tool_calls": query["tool_calls"]} + if query.get("tool_calls") + else {} + ), } for query in conversation.get("queries", []) ] @@ -1326,6 +1341,7 @@ class StreamProcessor: self, tool_actions: list, conversation_id: str, + claimed_state: Optional[Dict[str, Any]] = None, ): """Resume a paused agent from saved continuation state. @@ -1369,24 +1385,12 @@ class StreamProcessor: self.decoded_token = {"sub": owner} cont_service = ContinuationService() - state = cont_service.load_state(conversation_id, self.initial_user_id) + state = claimed_state or cont_service.claim_state( + conversation_id, self.initial_user_id + ) if not state: raise ValueError("No pending tool state found for this conversation") - # Claim the resume up-front. ``mark_resuming`` only flips ``pending`` - # → ``resuming``; if it returns False, another resume already - # claimed this row (status='resuming') — bail before any further - # LLM/tool work to avoid double-execution. The cleanup janitor - # reverts a stale ``resuming`` claim back to ``pending`` after the - # 10-minute grace window so the user can retry. - if not cont_service.mark_resuming( - conversation_id, self.initial_user_id, - ): - raise ValueError( - "Resume already in progress for this conversation; " - "retry after the grace window if it stalls." - ) - messages = state["messages"] pending_tool_calls = state["pending_tool_calls"] tools_dict = state["tools_dict"] @@ -1418,6 +1422,9 @@ class StreamProcessor: agent_id=agent_id, model_user_id=model_user_id, ) + importer = getattr(llm, "import_responses_state", None) + if callable(importer): + importer(agent_config.get("responses_state")) llm_handler = LLMHandlerCreator.create_handler(llm_name or "default") tool_executor = ToolExecutor( user_api_key=user_api_key, diff --git a/application/api/user/models/routes.py b/application/api/user/models/routes.py index afc80678..9bd5c5ca 100644 --- a/application/api/user/models/routes.py +++ b/application/api/user/models/routes.py @@ -56,6 +56,8 @@ def _user_id_or_401(): def _normalize_capabilities(raw) -> dict: """Coerce + bound the user-supplied capabilities payload.""" raw = raw or {} + if not isinstance(raw, dict): + raise ValueError("'capabilities' must be an object") out = {} if "supports_tools" in raw: out["supports_tools"] = bool(raw["supports_tools"]) @@ -97,6 +99,26 @@ def _normalize_capabilities(raw) -> dict: f"{_CONTEXT_WINDOW_MIN} and {_CONTEXT_WINDOW_MAX}" ) out["context_window"] = cw + if "api_flavor" in raw: + from application.core.model_yaml import VALID_API_FLAVORS + + api_flavor = raw["api_flavor"] + if api_flavor not in VALID_API_FLAVORS: + valid = ", ".join(sorted(VALID_API_FLAVORS)) + raise ValueError( + f"'capabilities.api_flavor' must be one of [{valid}]" + ) + out["api_flavor"] = api_flavor + if "reasoning_effort" in raw and raw["reasoning_effort"] is not None: + from application.core.model_yaml import VALID_REASONING_EFFORTS + + reasoning_effort = raw["reasoning_effort"] + if reasoning_effort not in VALID_REASONING_EFFORTS: + valid = ", ".join(sorted(VALID_REASONING_EFFORTS)) + raise ValueError( + f"'capabilities.reasoning_effort' must be one of [{valid}]" + ) + out["reasoning_effort"] = reasoning_effort return out @@ -360,21 +382,34 @@ class UserModelResource(Resource): def _run_connection_test( - base_url: str, api_key: str, upstream_model_id: str + base_url: str, + api_key: str, + upstream_model_id: str, + api_flavor: str = "chat_completions", ): - """Send a 1-token chat-completion to verify a BYOM endpoint. + """Send a minimal generation request to verify a BYOM endpoint. Returns ``(body, http_status)``. Upstream errors return 200 with ``ok=False`` so the UI can render inline errors; only local SSRF rejection returns 400. """ - url = base_url.rstrip("/") + "/chat/completions" - payload = { - "model": upstream_model_id, - "messages": [{"role": "user", "content": "hi"}], - "max_tokens": 1, - "stream": False, - } + if api_flavor == "responses": + url = base_url.rstrip("/") + "/responses" + payload = { + "model": upstream_model_id, + "input": "hi", + "max_output_tokens": 16, + "stream": False, + "store": False, + } + else: + url = base_url.rstrip("/") + "/chat/completions" + payload = { + "model": upstream_model_id, + "messages": [{"role": "user", "content": "hi"}], + "max_tokens": 1, + "stream": False, + } headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", @@ -427,8 +462,8 @@ class UserModelTestPayloadResource(Resource): "Test an arbitrary BYOM payload (display_name / model id / " "base_url / api_key) without saving. Used by the UI's 'Test " "connection' button so the user can validate before they " - "Save. Same SSRF guard, same 1-token request, same 5s " - "timeout as the by-id variant." + "Save. Uses the selected API protocol with the same SSRF guard " + "and 5s timeout as the by-id variant." ) ) def post(self): @@ -443,8 +478,15 @@ class UserModelTestPayloadResource(Resource): if missing: return missing + try: + capabilities = _normalize_capabilities(data.get("capabilities")) + except ValueError as e: + return make_response(jsonify({"ok": False, "error": str(e)}), 400) body, status = _run_connection_test( - data["base_url"], data["api_key"], data["upstream_model_id"] + data["base_url"], + data["api_key"], + data["upstream_model_id"], + capabilities.get("api_flavor", "chat_completions"), ) return make_response(jsonify(body), status) @@ -515,7 +557,18 @@ class UserModelTestResource(Resource): upstream_model_id = ( override_upstream_model_id or row["upstream_model_id"] ) + try: + capabilities = ( + _normalize_capabilities(data["capabilities"]) + if "capabilities" in data + else (row.get("capabilities") or {}) + ) + except ValueError as e: + return make_response(jsonify({"ok": False, "error": str(e)}), 400) body, status = _run_connection_test( - base_url, api_key, upstream_model_id + base_url, + api_key, + upstream_model_id, + capabilities.get("api_flavor", "chat_completions"), ) return make_response(jsonify(body), status) diff --git a/application/api/user/tasks.py b/application/api/user/tasks.py index 00efb718..7b45c7cc 100644 --- a/application/api/user/tasks.py +++ b/application/api/user/tasks.py @@ -15,6 +15,8 @@ from application.worker import ( sync_worker, ) +logger = logging.getLogger(__name__) + # Shared decorator config for long-running, side-effecting tasks. ``acks_late`` # is also the celeryconfig default but stays explicit here so each task's @@ -545,6 +547,30 @@ def cleanup_pending_tool_state(self): conversation_id = row.get("conversation_id") if not user_id or not conversation_id: continue + agent_config = row.get("agent_config") or {} + reserved_message_id = ( + agent_config.get("reserved_message_id") + if isinstance(agent_config, dict) + else None + ) + if reserved_message_id: + try: + from application.api.answer.services.conversation_service import ( + ConversationService, + TERMINATED_RESPONSE_PLACEHOLDER, + ) + + ConversationService().finalize_message( + str(reserved_message_id), + TERMINATED_RESPONSE_PLACEHOLDER, + status="failed", + error=TimeoutError("Tool continuation expired before resume"), + ) + except Exception: + logger.exception( + "Failed to retire expired continuation message %s", + reserved_message_id, + ) publish_user_event( str(user_id), "tool.approval.cleared", diff --git a/application/api/v1/routes.py b/application/api/v1/routes.py index ca293439..61389d5e 100644 --- a/application/api/v1/routes.py +++ b/application/api/v1/routes.py @@ -16,14 +16,28 @@ from flask import Blueprint, jsonify, make_response, request, Response from application.api.answer.routes.base import BaseAnswerResource from application.api.answer.services.persistence_policy import resolve_persistence +from application.api.answer.services.continuation_service import ( + ContinuationService, + ResumeInProgressError, +) from application.api.answer.services.stream_processor import StreamProcessor from application.api.v1 import idempotency as v1_idempotency +from application.api.v1.session_store import ( + V1Session, + delete_conversation, + identify_session, + load_conversation, + save_conversation, +) from application.api.v1.translator import ( + StreamTranslationState, + make_usage_chunk, translate_request, translate_response, translate_stream_event, ) from application.storage.db.repositories.agents import AgentsRepository +from application.storage.db.repositories.conversations import ConversationsRepository from application.storage.db.session import db_readonly logger = logging.getLogger(__name__) @@ -56,6 +70,77 @@ def _get_model_name(agent: Optional[Dict], api_key: str) -> str: return api_key +def _invalid_request(message: str, code: Optional[str] = None) -> Response: + """Return an OpenAI-shaped invalid-request response.""" + return make_response( + jsonify({ + "error": { + "message": message, + "type": "invalid_request_error", + **({"code": code} if code else {}), + } + }), + 400, + ) + + +def _validate_request_options(data: Dict[str, Any], agent: Dict[str, Any]) -> Optional[Response]: + """Reject unsupported options instead of silently changing semantics.""" + # The API key, not the OpenAI ``model`` placeholder, selects the agent. + # Keep accepting arbitrary model strings as documented for compatibility + # with clients that require a locally configured model alias. + if data.get("n") not in (None, 1): + return _invalid_request("DocsGPT currently supports only n=1.") + if data.get("logprobs") not in (None, False): + return _invalid_request("logprobs is not supported by this endpoint.") + stream_options = data.get("stream_options") + if stream_options is not None and not isinstance(stream_options, dict): + return _invalid_request("stream_options must be an object.") + return None + + +def _conversation_belongs_to_agent( + conversation_id: str, user_id: str, agent_id: str +) -> bool: + """Return whether a conversation is accessible and bound to this agent.""" + if not conversation_id or not user_id or not agent_id: + return False + try: + with db_readonly() as conn: + conversation = ConversationsRepository(conn).get_any( + str(conversation_id), str(user_id) + ) + except Exception: + logger.warning("Failed to authorize v1 conversation", exc_info=True) + return False + return bool( + conversation + and conversation.get("agent_id") + and str(conversation["agent_id"]) == str(agent_id) + ) + + +def _response_usage(agent: Any) -> Dict[str, Any]: + """Return the upstream provider's latest OpenAI-shaped usage object.""" + usage = getattr(getattr(agent, "llm", None), "_last_usage", None) + if isinstance(usage, dict): + return usage + tokens = getattr(getattr(agent, "llm", None), "token_usage", {}) or {} + prompt = int(tokens.get("prompt_tokens", 0) or 0) + completion = int(tokens.get("generated_tokens", 0) or 0) + return { + "prompt_tokens": prompt, + "completion_tokens": completion, + "total_tokens": prompt + completion, + } + + +def _response_finish_reason(agent: Any) -> str: + """Return the upstream terminal reason understood by Chat Completions.""" + reason = getattr(getattr(agent, "llm", None), "_last_finish_reason", None) + return reason if reason in {"stop", "length"} else "stop" + + class _V1AnswerHelper(BaseAnswerResource): """Thin wrapper to access complete_stream / process_response_stream.""" pass @@ -80,6 +165,14 @@ def chat_completions(): is_stream = data.get("stream", False) agent_doc = _lookup_agent(api_key) + if not agent_doc: + return make_response( + jsonify({"error": {"message": "Invalid API key", "type": "auth_error"}}), + 401, + ) + options_error = _validate_request_options(data, agent_doc) + if options_error is not None: + return options_error model_name = _get_model_name(agent_doc, api_key) # ---- Layer-1 idempotency (opt-in, non-streaming only) ---- @@ -113,6 +206,21 @@ def chat_completions(): 400, ) + agent_id_value = str(agent_doc.get("id") or agent_doc.get("_id") or "") + client_session = identify_session(request.headers, data, agent_id_value) + explicit_conversation = ( + request.headers.get("X-DocsGPT-Conversation-ID") + or internal_data.get("conversation_id") + ) + conversation_from_session = False + if explicit_conversation: + internal_data["conversation_id"] = explicit_conversation + elif not internal_data.get("conversation_id"): + correlated_conversation = load_conversation(client_session) + if correlated_conversation: + internal_data["conversation_id"] = correlated_conversation + conversation_from_session = True + # Link decoded_token to the agent's owner so continuation state, # logs, and tool execution use the correct user identity. The PG # ``agents`` row exposes the owner via ``user_id`` (``user`` is the @@ -124,50 +232,83 @@ def chat_completions(): ) decoded_token = {"sub": agent_user or "api_key_user"} + conversation_id = internal_data.get("conversation_id") + if conversation_id and not _conversation_belongs_to_agent( + str(conversation_id), str(decoded_token["sub"]), agent_id_value + ): + if conversation_from_session: + # A deleted or reassigned conversation must not poison this client + # session for the remainder of its Redis TTL. Start a fresh hidden + # conversation and replace the mapping after the request succeeds. + delete_conversation(client_session) + internal_data.pop("conversation_id", None) + else: + return _invalid_request( + "Conversation not found for the authenticated agent.", + code="conversation_not_found", + ) + + if internal_data.get("tool_actions") and internal_data.get("conversation_id"): + internal_data["persist"] = True + try: processor = StreamProcessor(internal_data, decoded_token) if internal_data.get("tool_actions"): - # Continuation mode — coherent Option B: the v1 tool round-trip is - # fully stateless. The pause finalized the prior turn's row as - # ``complete`` and wrote NO ``pending_tool_state`` (see - # ``complete_stream(finalize_tool_pause_as_complete=True)``), so we - # ALWAYS rebuild the agent + pending calls from the re-POSTed - # message history — even when the client threads back the - # ``conversation_id`` it got from the first response. - # - # We deliberately do NOT call ``resume_from_tool_actions`` here: - # its ``load_state`` would find no pending state and raise (→ HTTP - # 400), since OpenAI clients resume statelessly rather than via a - # native resume. ``resume_from_tool_actions`` stays in place for - # the native ``/stream`` + ``/api/answer`` routes, which are - # unchanged. conversation_id = internal_data.get("conversation_id") - ( - agent, - messages, - tools_dict, - pending_tool_calls, - tool_actions, - reasoning_content, - ) = processor.build_continuation_from_messages( - internal_data.get("messages", []), - internal_data["tool_actions"], + pending_state = ( + ContinuationService().claim_state( + conversation_id, decoded_token["sub"] + ) + if conversation_id + else None ) - # When a conversation_id is carried, target it for persistence so - # the final answer appends as a NEW terminal turn in that - # conversation (``save_conversation`` keys off ``conversation_id``) - # rather than creating an orphan sibling. ``build_continuation_from_messages`` - # leaves the processor's ``conversation_id`` (set from the request - # in ``__init__``) intact; pin it explicitly for clarity. - if conversation_id: + if conversation_id and pending_state: + ( + agent, + messages, + tools_dict, + pending_tool_calls, + tool_actions, + reasoning_content, + ) = processor.resume_from_tool_actions( + internal_data["tool_actions"], + conversation_id, + claimed_state=pending_state, + ) processor.conversation_id = conversation_id + else: + # Compatibility fallback for old/completed conversations and + # clients that resend the full transcript without resumable + # server state. StreamProcessor still enforces conversation + # ownership while loading history. + ( + agent, + messages, + tools_dict, + pending_tool_calls, + tool_actions, + reasoning_content, + ) = processor.build_continuation_from_messages( + internal_data.get("messages", []), + internal_data["tool_actions"], + ) + # A missing/expired durable continuation has no reserved WAL + # row to finalize. Run it statelessly without appending a blank + # sibling turn to the mapped conversation. + internal_data["persist"] = False continuation = { "messages": messages, "tools_dict": tools_dict, "pending_tool_calls": pending_tool_calls, "tool_actions": tool_actions, "reasoning_content": reasoning_content, + # Stateful compatibility resumes must finalize the original + # WAL placeholder and keep its request attribution. Omitting + # these made OpenCode rounds append an orphan response while + # leaving the initial message permanently ``streaming``. + "reserved_message_id": processor.reserved_message_id, + "request_id": processor.request_id, } question = "" else: @@ -199,6 +340,9 @@ def chat_completions(): strip_reasoning_leak = bool( internal_data.get("json_schema") or internal_data.get("json_object") ) + finalize_stateless_tool_pause = bool( + client_session is None and not internal_data.get("conversation_id") + ) if is_stream: # Idempotency replay is NOT supported for streaming: there is no @@ -216,6 +360,9 @@ def chat_completions(): should_persist, visibility, strip_reasoning_leak, + bool((data.get("stream_options") or {}).get("include_usage")), + client_session, + finalize_stateless_tool_pause, ), mimetype="text/event-stream", headers={ @@ -247,6 +394,8 @@ def chat_completions(): should_persist, visibility, strip_reasoning_leak, + client_session, + finalize_stateless_tool_pause, ) # Cache only successful (2xx) responses; ``finalize`` releases the @@ -255,6 +404,19 @@ def chat_completions(): v1_idempotency.finalize(idem_key, response) return response + except ResumeInProgressError as e: + if idem_key: + v1_idempotency.release(idem_key) + return make_response( + jsonify({ + "error": { + "message": str(e), + "type": "conflict_error", + "code": "resume_in_progress", + } + }), + 409, + ) except ValueError as e: if idem_key: v1_idempotency.release(idem_key) @@ -289,6 +451,9 @@ def _stream_response( should_persist: bool, visibility: str, strip_reasoning_leak: bool = False, + include_usage: bool = False, + client_session: Optional[V1Session] = None, + finalize_stateless_tool_pause: bool = False, ) -> Generator[str, None, None]: """Generate translated SSE chunks for streaming response.""" completion_id = f"chatcmpl-{int(time.time())}" @@ -305,12 +470,11 @@ def _stream_response( should_persist=should_persist, visibility=visibility, _continuation=continuation, - # OpenAI clients resume tool calls statelessly (no slot for our - # reserved_message_id), so a tool pause must finalize the row as - # ``complete`` here rather than stranding it for a native resume. - finalize_tool_pause_as_complete=True, + finalize_tool_pause_as_complete=finalize_stateless_tool_pause, ) + translation_state = StreamTranslationState() + for line in internal_stream: if not line.strip(): continue @@ -339,10 +503,19 @@ def _stream_response( conv_id = event_data.get("id", "") if conv_id and conv_id != "None": completion_id = f"chatcmpl-{conv_id}" + save_conversation(client_session, conv_id) # Translate to standard format + if event_data.get("type") == "end": + event_data["finish_reason"] = _response_finish_reason(agent) + if event_data.get("type") == "end" and include_usage: + yield make_usage_chunk(completion_id, model_name, _response_usage(agent)) translated = translate_stream_event( - event_data, completion_id, model_name, strip_reasoning_leak + event_data, + completion_id, + model_name, + strip_reasoning_leak, + translation_state, ) for chunk in translated: yield chunk @@ -358,6 +531,8 @@ def _non_stream_response( should_persist: bool, visibility: str, strip_reasoning_leak: bool = False, + client_session: Optional[V1Session] = None, + finalize_stateless_tool_pause: bool = False, ) -> Response: """Collect full response and return as single JSON.""" stream = helper.complete_stream( @@ -372,10 +547,7 @@ def _non_stream_response( should_persist=should_persist, visibility=visibility, _continuation=continuation, - # OpenAI clients resume tool calls statelessly (no slot for our - # reserved_message_id), so a tool pause must finalize the row as - # ``complete`` here rather than stranding it for a native resume. - finalize_tool_pause_as_complete=True, + finalize_tool_pause_as_complete=finalize_stateless_tool_pause, ) result = helper.process_response_stream(stream) @@ -388,6 +560,7 @@ def _non_stream_response( extra = result.get("extra") pending = extra.get("pending_tool_calls") if isinstance(extra, dict) else None + save_conversation(client_session, result.get("conversation_id")) response = translate_response( conversation_id=result["conversation_id"], @@ -398,6 +571,8 @@ def _non_stream_response( model_name=model_name, pending_tool_calls=pending, strip_reasoning_leak=strip_reasoning_leak, + usage=_response_usage(agent), + finish_reason_override=_response_finish_reason(agent), ) return make_response(jsonify(response), 200) diff --git a/application/api/v1/session_store.py b/application/api/v1/session_store.py new file mode 100644 index 00000000..cfd33768 --- /dev/null +++ b/application/api/v1/session_store.py @@ -0,0 +1,99 @@ +"""Short-lived correlation for chat-completions clients. + +OpenAI's chat-completions schema has no conversation id. Coding clients such +as OpenCode do send stable session headers, though, so map those opaque values +to DocsGPT conversations without storing or logging the raw client id. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +from dataclasses import dataclass +from typing import Any, Mapping, Optional + +from application.cache import get_redis_instance +from application.core.settings import settings + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class V1Session: + """A hashed Redis key for one agent/client/prompt namespace.""" + + key: str + + +def _instruction_fingerprint(messages: list[dict[str, Any]]) -> str: + instructions = [ + {"role": item.get("role"), "content": item.get("content")} + for item in messages + if item.get("role") in ("system", "developer") + ] + encoded = json.dumps(instructions, sort_keys=True, separators=(",", ":"), default=str) + return hashlib.sha256(encoded.encode("utf-8")).hexdigest() + + +def identify_session( + headers: Mapping[str, str], data: dict[str, Any], agent_id: str +) -> Optional[V1Session]: + """Build a privacy-preserving correlation key when the client supplies one.""" + raw = ( + headers.get("X-DocsGPT-Session-ID") + or (data.get("docsgpt") or {}).get("session_id") + or headers.get("X-Session-ID") + or headers.get("X-Session-Id") + or headers.get("X-Session-Affinity") + ) + if not raw: + return None + prompt = _instruction_fingerprint(data.get("messages") or []) + digest = hashlib.sha256(f"{agent_id}\0{raw}\0{prompt}".encode("utf-8")).hexdigest() + return V1Session(key=f"v1:session:{digest}") + + +def load_conversation(session: Optional[V1Session]) -> Optional[str]: + """Resolve a session key, degrading cleanly when Redis is unavailable.""" + if session is None: + return None + client = get_redis_instance() + if client is None: + return None + try: + value = client.get(session.key) + return value.decode("utf-8") if isinstance(value, bytes) else value + except Exception: + logger.warning("Unable to load v1 session correlation", exc_info=True) + return None + + +def save_conversation(session: Optional[V1Session], conversation_id: Optional[str]) -> None: + """Store a session mapping with a bounded lifetime.""" + if session is None or not conversation_id or conversation_id == "None": + return + client = get_redis_instance() + if client is None: + return + try: + client.set( + session.key, + str(conversation_id), + ex=settings.V1_SESSION_TTL_SECONDS, + ) + except Exception: + logger.warning("Unable to save v1 session correlation", exc_info=True) + + +def delete_conversation(session: Optional[V1Session]) -> None: + """Delete a stale session mapping, degrading cleanly without Redis.""" + if session is None: + return + client = get_redis_instance() + if client is None: + return + try: + client.delete(session.key) + except Exception: + logger.warning("Unable to delete v1 session correlation", exc_info=True) diff --git a/application/api/v1/translator.py b/application/api/v1/translator.py index 7ffc9325..14e28d48 100644 --- a/application/api/v1/translator.py +++ b/application/api/v1/translator.py @@ -9,6 +9,7 @@ This module handles: import json import re import time +from dataclasses import dataclass, field from typing import Any, Dict, List, Optional # Some upstream models/proxies echo their reasoning into ``content`` as @@ -143,14 +144,18 @@ def content_to_text(content: Any) -> str: def extract_system_prompt(messages: List[Dict]) -> Optional[str]: - """Extract the first system message content from the messages array. + """Combine system and developer instructions in request order. - Returns None if no system message is present. + Chat-completions clients use both roles. DocsGPT has one prompt-override + slot, so preserving their order is the least surprising translation. + Returns None when neither role is present. """ - for msg in messages: - if msg.get("role") == "system": - return content_to_text(msg.get("content", "")) - return None + prompts = [ + content_to_text(msg.get("content", "")) + for msg in messages + if msg.get("role") in ("system", "developer") + ] + return "\n\n".join(prompts) if prompts else None def convert_history(messages: List[Dict]) -> List[Dict]: @@ -163,7 +168,7 @@ def convert_history(messages: List[Dict]) -> List[Dict]: i = 0 while i < len(messages): msg = messages[i] - if msg.get("role") == "system": + if msg.get("role") in ("system", "developer"): i += 1 continue if msg.get("role") == "user": @@ -247,13 +252,19 @@ def translate_request( # max_completion_tokens, so drop the alias when the canonical key is present. if "max_completion_tokens" in sampling_params: sampling_params.pop("max_tokens", None) + if data.get("tools") and data.get("tool_choice") is not None: + sampling_params["tool_choice"] = data["tool_choice"] + if data.get("tools") and data.get("parallel_tool_calls") is not None: + sampling_params["parallel_tool_calls"] = data["parallel_tool_calls"] # Check for continuation (tool results after assistant tool_calls) if is_continuation(messages): tool_actions = extract_tool_results(messages) conversation_id = extract_conversation_id(messages) if not conversation_id: - conversation_id = data.get("conversation_id") + conversation_id = data.get("conversation_id") or ( + data.get("docsgpt") or {} + ).get("conversation_id") result = { "conversation_id": conversation_id, "tool_actions": tool_actions, @@ -263,6 +274,8 @@ def translate_request( # rebuilt from the resent messages instead of server-side state. "messages": messages, } + if data.get("conversation_id") and not result.get("conversation_id"): + result["conversation_id"] = data["conversation_id"] # Persistence: stateful continuations (carrying a conversation_id) # persist the final turn; stateless ones (no conversation_id, e.g. # opencode) skip it, else every tool round writes an orphan conversation @@ -307,6 +320,9 @@ def translate_request( # owner's sidebar; the legacy ``docsgpt.save_conversation`` flag # (old meaning: "persist this conversation") is ignored. } + conversation_id = data.get("conversation_id") or docsgpt.get("conversation_id") + if conversation_id: + result["conversation_id"] = conversation_id if system_prompt_override is not None: result["system_prompt_override"] = system_prompt_override @@ -346,6 +362,8 @@ def translate_response( model_name: str, pending_tool_calls: Optional[List[Dict]] = None, strip_reasoning_leak: bool = False, + usage: Optional[Dict[str, Any]] = None, + finish_reason_override: Optional[str] = None, ) -> Dict[str, Any]: """Translate DocsGPT response to chat completions format. @@ -395,7 +413,7 @@ def translate_response( combined_reasoning = (thought or "") + leaked_reasoning if combined_reasoning: message["reasoning_content"] = combined_reasoning - finish_reason = "stop" + finish_reason = finish_reason_override or "stop" result: Dict[str, Any] = { "id": completion_id, @@ -409,7 +427,7 @@ def translate_response( "finish_reason": finish_reason, } ], - "usage": { + "usage": usage or { "prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0, @@ -435,6 +453,15 @@ def translate_response( # --------------------------------------------------------------------------- +@dataclass +class StreamTranslationState: + """State shared while translating one internal stream.""" + + tool_indices: Dict[str, int] = field(default_factory=dict) + tool_pause_finished: bool = False + done: bool = False + + def _make_chunk( completion_id: str, model_name: str, @@ -458,6 +485,21 @@ def _make_chunk( return f"data: {json.dumps(chunk)}\n\n" +def make_usage_chunk( + completion_id: str, model_name: str, usage: Dict[str, Any] +) -> str: + """Build the optional final usage chunk used by ``stream_options``.""" + chunk = { + "id": completion_id, + "object": "chat.completion.chunk", + "created": int(time.time()), + "model": model_name, + "choices": [], + "usage": usage, + } + return f"data: {json.dumps(chunk)}\n\n" + + def _make_docsgpt_chunk(data: Dict[str, Any], completion_id: str, model_name: str) -> str: """Build a DocsGPT extension chunk that is ALSO a valid ``chat.completion.chunk``. @@ -483,6 +525,7 @@ def translate_stream_event( completion_id: str, model_name: str, strip_reasoning_leak: bool = False, + state: Optional[StreamTranslationState] = None, ) -> List[str]: """Translate a DocsGPT SSE event dict to standard streaming chunks. @@ -501,6 +544,7 @@ def translate_stream_event( """ event_type = event_data.get("type") chunks: List[str] = [] + state = state or StreamTranslationState() if event_type == "answer": raw = event_data.get("answer", "") @@ -540,11 +584,14 @@ def translate_stream_event( # Standard: stream as tool_calls delta args = tc_data.get("arguments", {}) args_str = json.dumps(args) if isinstance(args, dict) else str(args) + call_id = tc_data.get("call_id", "") + if call_id not in state.tool_indices: + state.tool_indices[call_id] = len(state.tool_indices) chunks.append( _make_chunk(completion_id, model_name, { "tool_calls": [{ - "index": 0, - "id": tc_data.get("call_id", ""), + "index": state.tool_indices[call_id], + "id": call_id, "type": "function", "function": { "name": _get_client_tool_name(tc_data), @@ -562,9 +609,11 @@ def translate_stream_event( elif event_type == "tool_calls_pending": # Standard: finish_reason = tool_calls - chunks.append( - _make_chunk(completion_id, model_name, {}, finish_reason="tool_calls") - ) + if not state.tool_pause_finished: + chunks.append( + _make_chunk(completion_id, model_name, {}, finish_reason="tool_calls") + ) + state.tool_pause_finished = True # Also emit as docsgpt extension chunks.append( _make_docsgpt_chunk( @@ -577,10 +626,18 @@ def translate_stream_event( ) elif event_type == "end": - chunks.append( - _make_chunk(completion_id, model_name, {}, finish_reason="stop") - ) - chunks.append("data: [DONE]\n\n") + if not state.done: + if not state.tool_pause_finished: + chunks.append( + _make_chunk( + completion_id, + model_name, + {}, + finish_reason=event_data.get("finish_reason") or "stop", + ) + ) + chunks.append("data: [DONE]\n\n") + state.done = True elif event_type == "id": # Skip the "None" placeholder conversation_id emitted when the call is diff --git a/application/cache.py b/application/cache.py index e5a82db0..99f738e4 100644 --- a/application/cache.py +++ b/application/cache.py @@ -176,23 +176,52 @@ def stream_cache(func): try: cached_response = redis_client.get(cache_key) if cached_response: - logger.info(f"Cache hit for stream key: {cache_key}") - cached_response = json.loads(cached_response.decode("utf-8")) - for chunk in cached_response: - yield chunk - time.sleep(0.03) # Simulate streaming delay - return + decoded = json.loads(cached_response.decode("utf-8")) + if ( + isinstance(decoded, dict) + and decoded.get("version") == 1 + and isinstance(decoded.get("chunks"), list) + ): + cached_chunks = decoded["chunks"] + elif isinstance(decoded, list) and not any( + isinstance(chunk, str) + and "_RespChoice" in chunk + for chunk in decoded + ): + # Backward-compatible read for pre-v1 string-only + # entries. Protocol-object reprs are deliberately + # rejected and refreshed from upstream. + cached_chunks = decoded + else: + cached_chunks = None + + if cached_chunks is not None: + logger.info(f"Cache hit for stream key: {cache_key}") + for chunk in cached_chunks: + yield chunk + time.sleep(0.03) # Simulate streaming delay + return + redis_client.delete(cache_key) except Exception as e: logger.error(f"Error getting cached stream: {e}", exc_info=True) stream_cache_data = [] + cacheable = True for chunk in func(self, model, messages, stream, tools, *args, **kwargs): yield chunk - stream_cache_data.append(str(chunk)) + if isinstance(chunk, (str, dict, list, int, float, bool, type(None))): + try: + json.dumps(chunk) + stream_cache_data.append(chunk) + except (TypeError, ValueError): + cacheable = False + else: + cacheable = False - if redis_client: + if redis_client and cacheable: try: - redis_client.set(cache_key, json.dumps(stream_cache_data), ex=1800) + payload = {"version": 1, "chunks": stream_cache_data} + redis_client.set(cache_key, json.dumps(payload), ex=1800) logger.info(f"Stream cache saved for key: {cache_key}") except Exception as e: logger.error(f"Error setting stream cache: {e}", exc_info=True) diff --git a/application/core/model_registry.py b/application/core/model_registry.py index 52eac732..10a1f010 100644 --- a/application/core/model_registry.py +++ b/application/core/model_registry.py @@ -322,6 +322,10 @@ class ModelRegistry: context_window=int( caps_raw.get("context_window") or 128000 ), + reasoning_effort=caps_raw.get("reasoning_effort"), + api_flavor=caps_raw.get( + "api_flavor", "chat_completions" + ), ) model_id = str(row["id"]) layer[model_id] = AvailableModel( diff --git a/application/core/model_settings.py b/application/core/model_settings.py index 654fbe64..19fe0318 100644 --- a/application/core/model_settings.py +++ b/application/core/model_settings.py @@ -78,6 +78,8 @@ class AvailableModel: "supports_structured_output": self.capabilities.supports_structured_output, "supports_streaming": self.capabilities.supports_streaming, "context_window": self.capabilities.context_window, + "api_flavor": self.capabilities.api_flavor, + "reasoning_effort": self.capabilities.reasoning_effort, "enabled": self.enabled, "source": self.source, } diff --git a/application/core/models/README.md b/application/core/models/README.md index 6d45323f..d72bd98a 100644 --- a/application/core/models/README.md +++ b/application/core/models/README.md @@ -111,6 +111,7 @@ defaults: # optional, applied to every model below input_cost_per_token: float # default null output_cost_per_token: float # default null reasoning_effort: # default null; none|minimal|low|medium|high|xhigh (subset is model-dependent) + api_flavor: # chat_completions (default) or responses models: # required - id: # unique registry key; persisted in agent records diff --git a/application/core/settings.py b/application/core/settings.py index 37a01eec..ce0230af 100644 --- a/application/core/settings.py +++ b/application/core/settings.py @@ -249,6 +249,15 @@ class Settings(BaseSettings): # reasoning items instead. OPENAI_RESPONSES_STORE: bool = False + # OpenAI-compatible clients can identify a logical chat with session + # headers even though chat-completions itself has no conversation field. + V1_SESSION_TTL_SECONDS: int = 24 * 60 * 60 + + # Optional cheaper model for first-party conversation titles. When unset, + # listed conversations use their answer model, but title work is still + # dispatched off the response path. + TITLE_MODEL_ID: Optional[str] = None + # Config-free tools on by default in agentless chats. ``scheduler`` is # dual-registered (also in ``BUILTIN_AGENT_TOOLS``) so the same synthetic id DEFAULT_CHAT_TOOLS: list = [ diff --git a/application/llm/openai.py b/application/llm/openai.py index 1ae7c070..b0b6b65d 100644 --- a/application/llm/openai.py +++ b/application/llm/openai.py @@ -1,4 +1,5 @@ import base64 +import hashlib import json import logging @@ -135,6 +136,7 @@ class OpenAILLM(BaseLLM): effective_base_url = settings.OPENAI_BASE_URL else: effective_base_url = "https://api.openai.com/v1" + self._effective_base_url = effective_base_url # http_client (set by LLMCreator for BYOM) is a DNS-rebinding-safe # httpx.Client; without it the SDK re-resolves DNS per request. @@ -156,6 +158,64 @@ class OpenAILLM(BaseLLM): # chain turns when OPENAI_RESPONSES_STORE is enabled. self._reasoning_for_calls = {} self._last_response_id = None + self._last_reasoning_items = [] + self._last_usage = None + self._imported_response_id = None + + def responses_chain_key(self) -> str: + """Return a credential- and endpoint-scoped Responses chain key. + + The digest is safe to persist in conversation metadata and prevents a + ``previous_response_id`` from being reused after the user switches + model, endpoint, or API credential. + + Returns: + A stable hexadecimal digest for the current Responses target. + """ + canonical_model_id = ( + getattr(self, "_canonical_model_id", None) or self.model_id or "" + ) + material = "\0".join( + ( + self.provider_name, + canonical_model_id, + self._effective_base_url, + self.api_key or "", + "store=true" if settings.OPENAI_RESPONSES_STORE else "store=false", + ) + ) + return hashlib.sha256(material.encode("utf-8")).hexdigest() + + def export_responses_state(self) -> dict: + """Return serializable Responses continuity state for persistence.""" + return { + "version": 1, + "chain_key": self.responses_chain_key(), + "response_id": ( + self._last_response_id if settings.OPENAI_RESPONSES_STORE else None + ), + "reasoning_items": self._last_reasoning_items, + "reasoning_for_calls": self._reasoning_for_calls, + } + + def import_responses_state(self, state: dict | None) -> bool: + """Restore encrypted/stored Responses state when its target matches.""" + if not isinstance(state, dict): + return False + if state.get("chain_key") != self.responses_chain_key(): + return False + self._imported_response_id = state.get("response_id") + self._last_reasoning_items = list(state.get("reasoning_items") or []) + self._reasoning_for_calls = dict(state.get("reasoning_for_calls") or {}) + return True + + def start_responses_turn(self) -> None: + """Reset continuity accumulated during the preceding user turn.""" + self._reasoning_for_calls = {} + self._last_reasoning_items = [] + self._last_response_id = None + self._imported_response_id = None + self._last_finish_reason = None def _clean_messages_openai(self, messages): cleaned_messages = [] @@ -197,6 +257,12 @@ class OpenAILLM(BaseLLM): } if reasoning_content: cleaned_assistant["reasoning_content"] = reasoning_content + if self._uses_responses_api() and message.get( + "responses_reasoning_items" + ): + cleaned_assistant["responses_reasoning_items"] = message[ + "responses_reasoning_items" + ] cleaned_messages.append(cleaned_assistant) continue @@ -215,6 +281,14 @@ class OpenAILLM(BaseLLM): msg_obj: dict = {"role": role, "content": content} if reasoning_content and role == "assistant": msg_obj["reasoning_content"] = reasoning_content + if ( + self._uses_responses_api() + and message.get("responses_reasoning_items") + and role == "assistant" + ): + msg_obj["responses_reasoning_items"] = message[ + "responses_reasoning_items" + ] cleaned_messages.append(msg_obj) elif isinstance(content, list): content_parts = [] @@ -324,7 +398,11 @@ class OpenAILLM(BaseLLM): **kwargs, ): messages = self._clean_messages_openai(messages) - logging.info(f"Cleaned messages: {_truncate_base64_for_logging(messages)}") + logging.debug( + "Prepared OpenAI request with %d messages and %d tools", + len(messages or []), + len(tools or []), + ) # Convert max_tokens to max_completion_tokens for newer models if "max_tokens" in kwargs: @@ -336,6 +414,9 @@ class OpenAILLM(BaseLLM): tools = None if response_format and not self._supports_structured_output(): response_format = None + if not tools: + kwargs.pop("tool_choice", None) + kwargs.pop("parallel_tool_calls", None) previous_response_id = kwargs.pop("previous_response_id", None) if self._uses_responses_api(): @@ -362,7 +443,7 @@ class OpenAILLM(BaseLLM): if response_format: request_params["response_format"] = response_format response = self.client.chat.completions.create(**request_params) - logging.info(f"OpenAI response: {response}") + logging.debug("OpenAI request completed") if tools: return response.choices[0] else: @@ -380,7 +461,11 @@ class OpenAILLM(BaseLLM): **kwargs, ): messages = self._clean_messages_openai(messages) - logging.info(f"Cleaned messages: {_truncate_base64_for_logging(messages)}") + logging.debug( + "Prepared OpenAI streaming request with %d messages and %d tools", + len(messages or []), + len(tools or []), + ) # Convert max_tokens to max_completion_tokens for newer models if "max_tokens" in kwargs: @@ -392,6 +477,9 @@ class OpenAILLM(BaseLLM): tools = None if response_format and not self._supports_structured_output(): response_format = None + if not tools: + kwargs.pop("tool_choice", None) + kwargs.pop("parallel_tool_calls", None) previous_response_id = kwargs.pop("previous_response_id", None) if self._uses_responses_api(): @@ -510,6 +598,14 @@ class OpenAILLM(BaseLLM): emitted_reasoning = set() for message in messages: role = message.get("role") + message_reasoning = message.get("responses_reasoning_items") or [] + for item in message_reasoning: + item_id = item.get("id") if isinstance(item, dict) else None + if item_id and item_id in emitted_reasoning: + continue + if item_id: + emitted_reasoning.add(item_id) + input_items.append(item) tool_calls = message.get("tool_calls") if tool_calls and role == "assistant": for tc in tool_calls: @@ -554,9 +650,7 @@ class OpenAILLM(BaseLLM): last completed assistant response needs to be sent again.""" last_assistant = -1 for i, message in enumerate(messages): - if message.get("role") == "assistant" and not message.get( - "tool_calls" - ): + if message.get("role") == "assistant": last_assistant = i if last_assistant < 0: return messages @@ -645,6 +739,16 @@ class OpenAILLM(BaseLLM): if tools: params["tools"] = self._to_responses_tools(tools) + if kwargs.get("tool_choice") is not None: + choice = kwargs["tool_choice"] + if isinstance(choice, dict) and choice.get("type") == "function": + choice = { + "type": "function", + "name": (choice.get("function") or {}).get("name", ""), + } + params["tool_choice"] = choice + if kwargs.get("parallel_tool_calls") is not None: + params["parallel_tool_calls"] = bool(kwargs["parallel_tool_calls"]) store = bool(settings.OPENAI_RESPONSES_STORE) params["store"] = store @@ -682,6 +786,49 @@ class OpenAILLM(BaseLLM): rid = getattr(response, "id", None) if rid: self._last_response_id = rid + usage = getattr(response, "usage", None) + if usage is not None: + prompt = int(getattr(usage, "input_tokens", 0) or 0) + completion = int(getattr(usage, "output_tokens", 0) or 0) + input_details = getattr(usage, "input_tokens_details", None) + output_details = getattr(usage, "output_tokens_details", None) + result = { + "prompt_tokens": prompt, + "completion_tokens": completion, + "total_tokens": int(getattr(usage, "total_tokens", 0) or prompt + completion), + } + cached = int(getattr(input_details, "cached_tokens", 0) or 0) + reasoning = int(getattr(output_details, "reasoning_tokens", 0) or 0) + if cached: + result["prompt_tokens_details"] = {"cached_tokens": cached} + if reasoning: + result["completion_tokens_details"] = {"reasoning_tokens": reasoning} + self._last_usage = result + + @staticmethod + def _responses_status_error(response) -> str | None: + """Return a terminal Responses error message, if one is present. + + Args: + response: An OpenAI Response object from a non-streaming call or + terminal streaming event. + + Returns: + A human-readable error string for failed/incomplete responses, or + ``None`` for successful and legacy status-less test objects. + """ + if response is None: + return None + status = getattr(response, "status", None) + error = getattr(response, "error", None) + if status == "failed" or error is not None: + message = getattr(error, "message", None) or str(error or "unknown error") + return f"Responses API failed: {message}" + details = getattr(response, "incomplete_details", None) + if status == "incomplete" or details is not None: + reason = getattr(details, "reason", None) or "unknown reason" + return f"Responses API incomplete: {reason}" + return None def _remember_reasoning(self, tool_calls, reasoning_items): """Key captured reasoning items by each function-call id for replay @@ -705,6 +852,8 @@ class OpenAILLM(BaseLLM): for part in getattr(item, "content", None) or []: if getattr(part, "type", None) == "output_text": content_parts.append(getattr(part, "text", "") or "") + elif getattr(part, "type", None) == "refusal": + content_parts.append(getattr(part, "refusal", "") or "") elif itype == "function_call": tool_calls.append(_RespToolCall( id=getattr(item, "call_id", "") or getattr(item, "id", ""), @@ -723,6 +872,12 @@ class OpenAILLM(BaseLLM): previous_response_id=None, **kwargs, ): + previous_response_id = ( + self._last_response_id or previous_response_id or self._imported_response_id + ) + self._last_response_id = None + self._last_usage = None + self._last_finish_reason = None if previous_response_id and settings.OPENAI_RESPONSES_STORE: messages = self._trim_for_previous_response(messages) input_items = self._to_responses_input(messages) @@ -736,11 +891,35 @@ class OpenAILLM(BaseLLM): kwargs=kwargs, ) response = self.client.responses.create(**params) - logging.info(f"OpenAI responses output: {getattr(response, 'output', None)}") + if response is None: + raise RuntimeError("Responses API returned no response object") + logging.debug( + "OpenAI Responses request completed id=%s", + getattr(response, "id", None), + ) self._record_responses_metadata(response) content, tool_calls, reasoning_items = self._parse_responses_output( response ) + self._last_reasoning_items = reasoning_items + details = getattr(response, "incomplete_details", None) + incomplete_reason = getattr(details, "reason", None) + if getattr(response, "status", None) == "incomplete": + if incomplete_reason != "max_output_tokens": + raise RuntimeError( + self._responses_status_error(response) + or "Responses API incomplete: unknown reason" + ) + self._last_finish_reason = "length" + if tools: + return _RespChoice( + finish_reason="length", + message=_RespMessage(content=content or None, tool_calls=None), + ) + return content or "" + status_error = self._responses_status_error(response) + if status_error: + raise RuntimeError(status_error) if tools: self._remember_reasoning(tool_calls, reasoning_items) message = _RespMessage( @@ -750,6 +929,7 @@ class OpenAILLM(BaseLLM): finish_reason="tool_calls" if tool_calls else "stop", message=message, ) + self._last_finish_reason = "stop" return content or "" def _responses_gen_stream( @@ -761,6 +941,12 @@ class OpenAILLM(BaseLLM): previous_response_id=None, **kwargs, ): + previous_response_id = ( + self._last_response_id or previous_response_id or self._imported_response_id + ) + self._last_response_id = None + self._last_usage = None + self._last_finish_reason = None if previous_response_id and settings.OPENAI_RESPONSES_STORE: messages = self._trim_for_previous_response(messages) input_items = self._to_responses_input(messages) @@ -777,6 +963,7 @@ class OpenAILLM(BaseLLM): func_calls = {} reasoning_items = [] + refusal_delta_seen = False try: for event in response: etype = getattr(event, "type", "") @@ -784,6 +971,15 @@ class OpenAILLM(BaseLLM): delta = getattr(event, "delta", "") if delta: yield delta + elif etype == "response.refusal.delta": + delta = getattr(event, "delta", "") + if delta: + refusal_delta_seen = True + yield delta + elif etype == "response.refusal.done" and not refusal_delta_seen: + refusal = getattr(event, "refusal", "") + if refusal: + yield refusal elif etype == "response.reasoning_summary_text.delta": delta = getattr(event, "delta", "") if delta: @@ -819,9 +1015,17 @@ class OpenAILLM(BaseLLM): self._reasoning_item_to_dict(item) ) elif etype == "response.completed": - self._record_responses_metadata( - getattr(event, "response", None) - ) + completed_response = getattr(event, "response", None) + if completed_response is None: + raise RuntimeError( + "Responses API returned no response object" + ) + status_error = self._responses_status_error(completed_response) + if status_error: + raise RuntimeError(status_error) + self._record_responses_metadata(completed_response) + self._last_reasoning_items = reasoning_items + self._last_finish_reason = "tool_calls" if func_calls else "stop" if func_calls: tool_calls = [] for position, index in enumerate(sorted(func_calls)): @@ -837,14 +1041,32 @@ class OpenAILLM(BaseLLM): finish_reason="tool_calls", delta=_RespDelta(tool_calls=tool_calls), ) + elif etype == "response.incomplete": + incomplete_response = getattr(event, "response", None) + details = getattr(incomplete_response, "incomplete_details", None) + reason = getattr(details, "reason", None) + if reason == "max_output_tokens": + self._record_responses_metadata(incomplete_response) + self._last_reasoning_items = reasoning_items + self._last_finish_reason = "length" + yield _RespChoice( + finish_reason="length", + delta=_RespDelta(tool_calls=None), + ) + return + status_error = self._responses_status_error(incomplete_response) + raise RuntimeError( + status_error or "Responses API incomplete: unknown reason" + ) elif etype in ("response.failed", "error"): resp = getattr(event, "response", None) - err = ( - getattr(resp, "error", None) - or getattr(event, "message", None) - or "Responses stream error" - ) - raise RuntimeError(f"Responses API stream error: {err}") + err = self._responses_status_error(resp) + if err is None: + err = ( + getattr(event, "message", None) + or "Responses API stream error" + ) + raise RuntimeError(err) finally: if hasattr(response, "close"): response.close() diff --git a/application/llm/providers/openai_compatible.py b/application/llm/providers/openai_compatible.py index 92f0b99a..e1db2959 100644 --- a/application/llm/providers/openai_compatible.py +++ b/application/llm/providers/openai_compatible.py @@ -145,5 +145,7 @@ class OpenAICompatibleProvider(Provider): enabled=model.enabled, base_url=model.base_url or base_url, display_provider=model.display_provider, + upstream_model_id=model.upstream_model_id, + source=model.source, api_key=api_key, ) diff --git a/application/storage/db/repositories/pending_tool_state.py b/application/storage/db/repositories/pending_tool_state.py index 06299215..5450e568 100644 --- a/application/storage/db/repositories/pending_tool_state.py +++ b/application/storage/db/repositories/pending_tool_state.py @@ -1,14 +1,14 @@ """Repository for the ``pending_tool_state`` table. -Mirrors the continuation service's three operations on -``pending_tool_state`` in Mongo: +Provides the continuation lifecycle operations for ``pending_tool_state``: - save_state → upsert (INSERT ... ON CONFLICT DO UPDATE) -- load_state → find_one by (conversation_id, user_id) +- load_state → fetch live pending state by (conversation_id, user_id) +- claim_state → atomically transition live pending state to ``resuming`` - delete_state → delete_one by (conversation_id, user_id) -Adds ``mark_resuming`` so a resumed run can claim a row without -deleting it; a separate ``revert_stale_resuming`` flips abandoned +Retains ``mark_resuming`` for compatibility; new resume paths use the atomic +claim. A separate ``revert_stale_resuming`` flips abandoned ``resuming`` rows back to ``pending`` so a crashed worker doesn't strand the user. @@ -102,6 +102,22 @@ class PendingToolStateRepository: return row_to_dict(result.fetchone()) def load_state(self, conversation_id: str, user_id: str) -> Optional[dict]: + """Load live pending state without exposing expired/resuming rows.""" + result = self._conn.execute( + text( + "SELECT * FROM pending_tool_state " + "WHERE conversation_id = CAST(:conv_id AS uuid) " + "AND user_id = :user_id " + "AND status = 'pending' " + "AND expires_at > clock_timestamp()" + ), + {"conv_id": conversation_id, "user_id": user_id}, + ) + row = result.fetchone() + return row_to_dict(row) if row is not None else None + + def load_state_any(self, conversation_id: str, user_id: str) -> Optional[dict]: + """Load state regardless of lifecycle for conflict classification.""" result = self._conn.execute( text( "SELECT * FROM pending_tool_state " @@ -113,6 +129,25 @@ class PendingToolStateRepository: row = result.fetchone() return row_to_dict(row) if row is not None else None + def claim_state(self, conversation_id: str, user_id: str) -> Optional[dict]: + """Atomically claim one live pending continuation and return it.""" + result = self._conn.execute( + text( + """ + UPDATE pending_tool_state + SET status = 'resuming', resumed_at = clock_timestamp() + WHERE conversation_id = CAST(:conv_id AS uuid) + AND user_id = :user_id + AND status = 'pending' + AND expires_at > clock_timestamp() + RETURNING * + """ + ), + {"conv_id": conversation_id, "user_id": user_id}, + ) + row = result.fetchone() + return row_to_dict(row) if row is not None else None + def delete_state(self, conversation_id: str, user_id: str) -> bool: result = self._conn.execute( text( @@ -134,6 +169,7 @@ class PendingToolStateRepository: WHERE conversation_id = CAST(:conv_id AS uuid) AND user_id = :user_id AND status = 'pending' + AND expires_at > clock_timestamp() """ ), {"conv_id": conversation_id, "user_id": user_id}, @@ -177,7 +213,7 @@ class PendingToolStateRepository: result = self._conn.execute( text( "DELETE FROM pending_tool_state WHERE expires_at < clock_timestamp() " - "RETURNING conversation_id, user_id" + "RETURNING conversation_id, user_id, agent_config" ) ) return [row_to_dict(r) for r in result.fetchall()] diff --git a/application/storage/db/repositories/user_custom_models.py b/application/storage/db/repositories/user_custom_models.py index a6ad3e02..38ab9f03 100644 --- a/application/storage/db/repositories/user_custom_models.py +++ b/application/storage/db/repositories/user_custom_models.py @@ -34,6 +34,8 @@ _ALLOWED_CAPABILITY_KEYS = frozenset( "supports_streaming", "attachments", "context_window", + "api_flavor", + "reasoning_effort", } ) diff --git a/application/usage.py b/application/usage.py index 9ae875fa..2e625031 100644 --- a/application/usage.py +++ b/application/usage.py @@ -140,6 +140,21 @@ def _persist_call_usage(llm, call_usage): logger.exception("token_usage persist failed") +def _prefer_provider_usage(llm, call_usage): + """Replace estimates with upstream counts when a provider reported them.""" + reported = getattr(llm, "_last_usage", None) + if not isinstance(reported, dict): + return call_usage + prompt = reported.get("prompt_tokens") + completion = reported.get("completion_tokens") + if prompt is None or completion is None: + return call_usage + return { + "prompt_tokens": int(prompt or 0), + "generated_tokens": int(completion or 0), + } + + def gen_token_usage(func): """Accumulate per-call token counts and write a ``token_usage`` row. @@ -172,6 +187,7 @@ def gen_token_usage(func): error = exc raise finally: + call_usage = _prefer_provider_usage(self, call_usage) self.token_usage["prompt_tokens"] += call_usage["prompt_tokens"] self.token_usage["generated_tokens"] += call_usage["generated_tokens"] _persist_call_usage(self, call_usage) @@ -219,6 +235,7 @@ def stream_token_usage(func): finally: for line in batch: call_usage["generated_tokens"] += _count_tokens(line) + call_usage = _prefer_provider_usage(self, call_usage) self.token_usage["prompt_tokens"] += call_usage["prompt_tokens"] self.token_usage["generated_tokens"] += call_usage["generated_tokens"] _persist_call_usage(self, call_usage) diff --git a/docs/content/Deploying/DocsGPT-Settings.mdx b/docs/content/Deploying/DocsGPT-Settings.mdx index 452ea6ed..160327df 100644 --- a/docs/content/Deploying/DocsGPT-Settings.mdx +++ b/docs/content/Deploying/DocsGPT-Settings.mdx @@ -431,6 +431,8 @@ See [Access Control, Roles & Teams](/Deploying/Access-Control) for the full mode | Setting | Default | Description | | --- | --- | --- | | `OPENAI_RESPONSES_STORE` | `false` | When `true`, allows OpenAI to persist [Responses API](/Models/cloud-providers#openai-responses-api-and-reasoning) state server-side. | +| `V1_SESSION_TTL_SECONDS` | `86400` | Redis correlation lifetime, in seconds, for OpenAI-compatible client sessions. Raw external session IDs are hashed and never stored. | +| `TITLE_MODEL_ID` | unset | Optional model ID used for asynchronous first-party conversation titles. When unset, the answer model is used. Hidden API conversations never invoke title generation. | ## Realtime Events Settings diff --git a/docs/content/Models/cloud-providers.mdx b/docs/content/Models/cloud-providers.mdx index 1ab730e1..b11973a8 100644 --- a/docs/content/Models/cloud-providers.mdx +++ b/docs/content/Models/cloud-providers.mdx @@ -67,10 +67,11 @@ This is the mechanism behind catalog entries like DeepSeek, which declare their For OpenAI models that support it, DocsGPT can call the newer **Responses API** (`/v1/responses`) instead of Chat Completions. This is selected per model in the catalog via an `api_flavor: responses` capability and enables features like server-side reasoning. Related settings: - `reasoning_effort` — per-model reasoning effort hint (for example `medium`) declared in the model catalog. -- `OPENAI_RESPONSES_STORE` (default `false`) — when `true`, lets OpenAI persist Responses API state server-side. +- `OPENAI_RESPONSES_STORE` (default `false`) — when `true`, lets OpenAI persist Responses API state server-side and chains calls with `previous_response_id`. When `false`, DocsGPT requests encrypted reasoning items and persists those ciphertext items itself so reasoning survives tool calls and later conversation turns without OpenAI retaining the response. +- `V1_SESSION_TTL_SECONDS` (default `86400`) — lifetime of the hashed Redis mapping used to associate OpenAI-compatible client session headers with hidden DocsGPT conversations. Explicit DocsGPT conversation IDs always take precedence. See [App Configuration](/Deploying/DocsGPT-Settings) for the full settings reference. ## Adding Support for Other Cloud Providers -If you wish to connect to a cloud provider that is not explicitly listed above or doesn't offer OpenAI API compatibility, you can extend DocsGPT to support it. Within the DocsGPT repository, navigate to the `application/llm` directory. Here, you will find Python files defining the existing LLM integrations. You can use these files as examples to create a new module for your desired cloud provider. After creating your new LLM module, you will need to register it within the `llm_creator.py` file. This process involves some coding, but it allows for virtually unlimited extensibility to connect to any cloud-based LLM service with an accessible API. \ No newline at end of file +If you wish to connect to a cloud provider that is not explicitly listed above or doesn't offer OpenAI API compatibility, you can extend DocsGPT to support it. Within the DocsGPT repository, navigate to the `application/llm` directory. Here, you will find Python files defining the existing LLM integrations. You can use these files as examples to create a new module for your desired cloud provider. After creating your new LLM module, you will need to register it within the `llm_creator.py` file. This process involves some coding, but it allows for virtually unlimited extensibility to connect to any cloud-based LLM service with an accessible API. diff --git a/frontend/src/api/services/customModelsService.ts b/frontend/src/api/services/customModelsService.ts index 37392681..e140413c 100644 --- a/frontend/src/api/services/customModelsService.ts +++ b/frontend/src/api/services/customModelsService.ts @@ -4,6 +4,7 @@ import endpoints from '../endpoints'; import type { CreateCustomModelPayload, CustomModel, + CustomModelCapabilities, CustomModelTestResult, } from '../../models/types'; @@ -84,6 +85,7 @@ const customModelsService = { base_url: string; api_key: string; upstream_model_id: string; + capabilities?: Pick; }, token: string | null, ): Promise => { @@ -122,14 +124,17 @@ const customModelsService = { base_url?: string; api_key?: string; upstream_model_id?: string; + capabilities?: Pick; } = {}, ): Promise => { // Send only non-empty overrides; server falls back to stored values. - const requestBody: Record = {}; + const requestBody: Record = {}; if (overrides.base_url) requestBody.base_url = overrides.base_url; if (overrides.api_key) requestBody.api_key = overrides.api_key; if (overrides.upstream_model_id) requestBody.upstream_model_id = overrides.upstream_model_id; + if (overrides.capabilities) + requestBody.capabilities = overrides.capabilities; const response = await apiClient.post( endpoints.USER.CUSTOM_MODEL_TEST(id), requestBody, diff --git a/frontend/src/locale/en.json b/frontend/src/locale/en.json index 2efb9eb0..d474b801 100644 --- a/frontend/src/locale/en.json +++ b/frontend/src/locale/en.json @@ -695,6 +695,15 @@ "capabilities": { "title": "Capabilities", "contextWindowShort": "Context window", + "apiFlavor": "API protocol", + "apiFlavors": { + "chatCompletions": "Chat Completions", + "responses": "Responses" + }, + "reasoningEffort": "Reasoning effort", + "reasoningEfforts": { + "default": "Provider default" + }, "chips": { "tools": "Tools", "structuredOutput": "Structured output", diff --git a/frontend/src/modals/CustomModelModal.tsx b/frontend/src/modals/CustomModelModal.tsx index b01c76ae..c217a616 100644 --- a/frontend/src/modals/CustomModelModal.tsx +++ b/frontend/src/modals/CustomModelModal.tsx @@ -8,11 +8,24 @@ import Spinner from '../components/Spinner'; import { Button } from '../components/ui/button'; import { Input } from '../components/ui/input'; import { Label } from '../components/ui/label'; +import { + Select, + SelectContent, + SelectItem, + SelectTrigger, + SelectValue, +} from '../components/ui/select'; import { ActiveState } from '../models/misc'; import { selectToken } from '../preferences/preferenceSlice'; import { Modal } from '../components/ui/modal'; -import type { CreateCustomModelPayload, CustomModel } from '../models/types'; +import type { + CreateCustomModelPayload, + CustomModel, + CustomModelCapabilities, + ModelApiFlavor, + ReasoningEffort, +} from '../models/types'; interface CustomModelModalProps { modalState: ActiveState; @@ -31,6 +44,8 @@ interface FormState { supports_structured_output: boolean; supports_images: boolean; context_window: number | ''; + api_flavor: ModelApiFlavor; + reasoning_effort: ReasoningEffort | 'default'; enabled: boolean; } @@ -50,6 +65,8 @@ const buildInitialFormState = (model?: CustomModel | null): FormState => { supports_structured_output: true, supports_images: false, context_window: DEFAULT_CONTEXT_WINDOW, + api_flavor: 'chat_completions', + reasoning_effort: 'default', enabled: true, }; } @@ -68,6 +85,8 @@ const buildInitialFormState = (model?: CustomModel | null): FormState => { supports_images: attachments.includes('image'), context_window: model.capabilities?.context_window ?? DEFAULT_CONTEXT_WINDOW, + api_flavor: model.capabilities?.api_flavor ?? 'chat_completions', + reasoning_effort: model.capabilities?.reasoning_effort ?? 'default', enabled: model.enabled ?? true, }; }; @@ -172,17 +191,22 @@ export default function CustomModelModal({ formData.context_window === '' ? DEFAULT_CONTEXT_WINDOW : Number(formData.context_window); + const capabilities: CustomModelCapabilities = { + supports_tools: formData.supports_tools, + supports_structured_output: formData.supports_structured_output, + attachments: formData.supports_images ? ['image'] : [], + context_window: ctxValue, + api_flavor: formData.api_flavor, + }; + if (formData.reasoning_effort !== 'default') { + capabilities.reasoning_effort = formData.reasoning_effort; + } const payload: CreateCustomModelPayload = { upstream_model_id: formData.upstream_model_id.trim(), display_name: formData.display_name.trim(), description: formData.description.trim(), base_url: formData.base_url.trim(), - capabilities: { - supports_tools: formData.supports_tools, - supports_structured_output: formData.supports_structured_output, - attachments: formData.supports_images ? ['image'] : [], - context_window: ctxValue, - }, + capabilities, enabled: formData.enabled, }; if (formData.api_key.trim()) { @@ -246,18 +270,23 @@ export default function CustomModelModal({ setTesting(true); setTestResult(null); try { + const capabilities = { + api_flavor: formData.api_flavor, + }; const result = isEditMode && model?.id ? await customModelsService.testCustomModel(model.id, token, { base_url: trimmedBaseUrl, api_key: trimmedApiKey, upstream_model_id: trimmedUpstreamId, + capabilities, }) : await customModelsService.testCustomModelPayload( { base_url: trimmedBaseUrl, api_key: trimmedApiKey, upstream_model_id: trimmedUpstreamId, + capabilities, }, token, ); @@ -447,6 +476,80 @@ export default function CustomModelModal({ {/* Row 4: Capabilities — flat (no border), chips + inline ctx */}
+
+
+ + +
+
+ + +
+
str: +def _seed_conversation( + conn, user_id: str, name: str, agent_id: str | None = None +) -> str: """Create a conversation owned by ``user_id`` with one finalized turn.""" repo = ConversationsRepository(conn) - conv = repo.create(user_id, name) + conv = repo.create(user_id, name, agent_id=agent_id) conv_id = str(conv["id"]) repo.reserve_message( conv_id, @@ -72,8 +58,10 @@ def _seed_conversation(conn, user_id: str, name: str) -> str: return conv_id -def _seed_agent(conn, user_id: str, key: str) -> None: - AgentsRepository(conn).create(user_id, "Weather Agent", "published", key=key) +def _seed_agent(conn, user_id: str, key: str) -> dict: + return AgentsRepository(conn).create( + user_id, "Weather Agent", "published", key=key + ) def _build_app() -> Flask: @@ -141,9 +129,11 @@ class TestForeignConversationOwnership: _seed_user(conn, user_a) _seed_user(conn, user_b) # Agent owned by B → the v1 route resolves decoded_token={"sub": B}. - _seed_agent(conn, user_b, api_key) + agent = _seed_agent(conn, user_b, api_key) # Conversation owned by A, with one existing complete turn. - conv_id = _seed_conversation(conn, user_a, "A's private chat") + conv_id = _seed_conversation( + conn, user_a, "A's private chat", agent_id=str(agent["id"]) + ) with pg_engine.connect() as conn: rows_before = _row_count(conn, conv_id) @@ -173,7 +163,7 @@ class TestForeignConversationOwnership: # The ownership gate raised ValueError → route returns 400. assert resp.status_code == 400, resp.get_data(as_text=True) body = resp.get_json() - assert body["error"]["type"] == "invalid_request" + assert body["error"]["type"] == "invalid_request_error" # No assistant turn was appended to A's conversation — the foreign # write was rejected before any persistence. @@ -191,9 +181,11 @@ class TestForeignConversationOwnership: with pg_engine.begin() as conn: _seed_user(conn, owner) - _seed_agent(conn, owner, api_key) + agent = _seed_agent(conn, owner, api_key) # Conversation owned by the agent owner (the caller). - conv_id = _seed_conversation(conn, owner, "owner's chat") + conv_id = _seed_conversation( + conn, owner, "owner's chat", agent_id=str(agent["id"]) + ) app = _build_app() diff --git a/tests/api/v1/test_v1_tool_pause_finalization.py b/tests/api/v1/test_v1_tool_pause_finalization.py index f1cd4502..bedc734b 100644 --- a/tests/api/v1/test_v1_tool_pause_finalization.py +++ b/tests/api/v1/test_v1_tool_pause_finalization.py @@ -586,23 +586,14 @@ class TestNativePauseUnchanged: # --------------------------------------------------------------------------- -# Guarantee 2 — coherent Option B routing: a v1 continuation carrying a -# conversation_id rebuilds STATELESSLY via build_continuation_from_messages -# (never resume_from_tool_actions), and threads no reserved_message_id. +# Guarantee 2 — legacy continuations without pending server state still fall +# back to a stateless transcript rebuild. # --------------------------------------------------------------------------- @pytest.mark.unit class TestV1ContinuationRoutesStatelessly: - """Coherent Option B: a v1 tool-result continuation is fully stateless on - the resume side. Because the pause finalized the prior row ``complete`` and - wrote NO ``pending_tool_state``, the route must rebuild from the re-POSTed - history via ``build_continuation_from_messages`` — **even when a - ``conversation_id`` is carried** — and must NOT call - ``resume_from_tool_actions`` (whose ``load_state`` would 400). The - ``_continuation`` dict therefore carries no live ``reserved_message_id`` / - ``request_id`` (id-reuse is inert under this design). - """ + """A carried conversation id with no pending state remains compatible.""" def _build_app(self) -> Flask: app = Flask(__name__) @@ -667,6 +658,15 @@ class TestV1ContinuationRoutesStatelessly: yield MagicMock() with patch( + "application.api.v1.routes._lookup_agent", + return_value={"id": "agent-1", "name": "Agent", "user_id": "owner"}, + ), patch( + "application.api.v1.routes.ContinuationService.claim_state", + return_value=None, + ), patch( + "application.api.v1.routes._conversation_belongs_to_agent", + return_value=True, + ), patch( "application.api.v1.routes.translate_request", side_effect=_fake_translate, ), patch( @@ -708,13 +708,12 @@ class TestV1ContinuationRoutesStatelessly: continuation = captured.get("_continuation") assert continuation is not None - # id-reuse is gone from the v1 continuation dict (inert under Option B). - assert "reserved_message_id" not in continuation - assert "request_id" not in continuation - # The v1 path still runs in stateless-finalize mode. - assert captured.get("finalize_tool_pause_as_complete") is True - # The incoming conversation_id is the persistence target. + assert continuation["reserved_message_id"] is None + assert continuation["request_id"] is None + assert captured.get("finalize_tool_pause_as_complete") is False + # Missing durable state runs without appending a blank sibling turn. assert captured.get("conversation_id") == "conv-1" + assert captured.get("should_persist") is False # --------------------------------------------------------------------------- @@ -737,6 +736,9 @@ class _FakeLLMGen: def gen(self, *args, **kwargs) -> str: return "Title" + def gen_stream(self, *args, **kwargs): + yield "It is 72F and sunny in SF." + class _FakeClientToolExecutor: """Tool executor stub exposing ``get_tools`` for the real @@ -912,6 +914,9 @@ class TestV1ToolRoundTripEndToEnd: "application.api.answer.services.stream_processor.StreamProcessor" ".build_agent", _fake_build_agent, + ), patch( + "application.api.v1.routes._conversation_belongs_to_agent", + return_value=True, ): with app.test_client() as c: resp1 = _post_chat( @@ -919,7 +924,10 @@ class TestV1ToolRoundTripEndToEnd: { "messages": [{"role": "user", "content": "weather in SF?"}], "tools": [self.CLIENT_TOOL], - "docsgpt": {"save_conversation": True}, + "docsgpt": { + "save_conversation": True, + "session_id": "tool-session", + }, }, api_key, ) @@ -936,19 +944,21 @@ class TestV1ToolRoundTripEndToEnd: conv_id = body1.get("docsgpt", {}).get("conversation_id") assert conv_id - # Reserved row finalized ``complete`` (with tool_calls); no - # ``pending_tool_state`` and no non-terminal row. + # The pause remains resumable and owns the original WAL placeholder. with pg_engine.connect() as conn: statuses = _row_statuses(conn, conv_id) - assert statuses == ["complete"] - assert _pending_tool_state_count(conn, conv_id) == 0 - assert self._count_non_terminal(conn, user_id) == 0 + assert statuses[0] in ("pending", "streaming") + assert _pending_tool_state_count(conn, conv_id) == 1 + assert self._count_non_terminal(conn, user_id) == 1 # ---- POST #2: tool result + conversation_id -> agent answers ---- with _wire_v1_route_db(pg_engine, monkeypatch), patch( "application.api.answer.services.stream_processor.StreamProcessor" ".build_agent", _fake_build_agent, + ), patch( + "application.api.v1.routes._conversation_belongs_to_agent", + return_value=True, ): with app.test_client() as c: resp2 = _post_chat( @@ -977,7 +987,10 @@ class TestV1ToolRoundTripEndToEnd: }, ], "conversation_id": conv_id, - "docsgpt": {"save_conversation": True}, + "docsgpt": { + "save_conversation": True, + "session_id": "tool-session", + }, }, api_key, ) @@ -996,12 +1009,12 @@ class TestV1ToolRoundTripEndToEnd: from sqlalchemy import text # The answer is a NEW terminal turn appended to the same - # conversation: the original tool-call turn + the answer turn. + # conversation: the original WAL placeholder is finalized. statuses = _row_statuses(conn, conv_id) - assert statuses == ["complete", "complete"] + assert statuses == ["complete"] # Nothing left non-terminal anywhere for this user / conversation. assert self._count_non_terminal(conn, user_id) == 0 - # The appended turn carries the assistant answer. + # The finalized turn carries the assistant answer. answer_rows = conn.execute( text( "SELECT response FROM conversation_messages " @@ -1017,4 +1030,3 @@ class TestV1ToolRoundTripEndToEnd: {"u": user_id}, ).scalar() assert conv_count == 1 - diff --git a/tests/core/test_byom_user_aware_helpers.py b/tests/core/test_byom_user_aware_helpers.py index 6e3c1a09..6ffde7ba 100644 --- a/tests/core/test_byom_user_aware_helpers.py +++ b/tests/core/test_byom_user_aware_helpers.py @@ -791,7 +791,7 @@ class TestContinuationPreservesByomScope: ) cont_service = MagicMock() - cont_service.load_state.return_value = { + cont_service.claim_state.return_value = { "messages": [], "pending_tool_calls": [], "tools_dict": {}, diff --git a/tests/core/test_model_registry_yaml.py b/tests/core/test_model_registry_yaml.py index df1105c3..791f2d19 100644 --- a/tests/core/test_model_registry_yaml.py +++ b/tests/core/test_model_registry_yaml.py @@ -90,8 +90,11 @@ def _make_settings(**overrides): def _reset_registry(monkeypatch): ModelRegistry.reset() # openai_compatible catalogs read their key directly from os.environ, - # so clear any host-set keys to keep these tests deterministic. - monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False) + # so clear every built-in/operator catalog key to keep these tests + # deterministic even when a developer has an ignored ``*_internal.yaml``. + for catalog in load_model_yamls([BUILTIN_MODELS_DIR]): + if catalog.api_key_env: + monkeypatch.delenv(catalog.api_key_env, raising=False) yield ModelRegistry.reset() @@ -118,7 +121,17 @@ class TestYAMLLoader: def test_each_provider_has_expected_ids(self): grouped = _by_provider(load_model_yamls([BUILTIN_MODELS_DIR])) for provider, expected in EXPECTED_IDS.items(): - actual = {m.id for c in grouped[provider] for m in c.models} + # Local ``*_internal.yaml`` files are intentionally gitignored + # operator extensions, not part of the canonical catalog snapshot. + canonical = [ + c + for c in grouped[provider] + if not ( + c.source_path + and c.source_path.stem.endswith("_internal") + ) + ] + actual = {m.id for c in canonical for m in c.models} assert actual == expected, f"{provider}: expected {expected}, got {actual}" def test_attachment_alias_image_expands_to_five_mime_types(self): diff --git a/tests/core/test_openai_compatible.py b/tests/core/test_openai_compatible.py index 4001197a..78270e5e 100644 --- a/tests/core/test_openai_compatible.py +++ b/tests/core/test_openai_compatible.py @@ -16,6 +16,7 @@ import pytest from application.core.model_registry import ModelRegistry from application.core.model_settings import ModelProvider +from application.core.model_yaml import BUILTIN_MODELS_DIR, load_model_yamls def _make_settings(**overrides): @@ -81,7 +82,9 @@ def _reset_registry(monkeypatch): # so a key leaked by another test (or present in the dev .env) doesn't add # extra models to the exact-match assertions below. Mirrors the delenv # guard in test_model_registry_yaml.py. - monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False) + for catalog in load_model_yamls([BUILTIN_MODELS_DIR]): + if catalog.api_key_env: + monkeypatch.delenv(catalog.api_key_env, raising=False) ModelRegistry.reset() yield ModelRegistry.reset() @@ -211,6 +214,33 @@ class TestYAMLCompatibleProvider: for v in d.values(): assert v != "sk" + def test_materialization_preserves_upstream_model_id( + self, tmp_path, monkeypatch + ): + path = _write_mistral_yaml(tmp_path) + path.write_text( + dedent(""" + provider: openai_compatible + display_provider: mistral + api_key_env: MISTRAL_API_KEY + base_url: https://api.mistral.ai/v1 + models: + - id: mistral-large-high + upstream_model_id: mistral-large-latest + display_name: Mistral Large (High) + """) + ) + monkeypatch.setenv("MISTRAL_API_KEY", "sk") + + s = _make_settings(MODELS_CONFIG_DIR=str(tmp_path)) + with patch("application.core.settings.settings", s): + reg = ModelRegistry() + + assert ( + reg.get_model("mistral-large-high").upstream_model_id + == "mistral-large-latest" + ) + # ── Legacy OPENAI_BASE_URL fallback ────────────────────────────────────── diff --git a/tests/core/test_registry_user_layer.py b/tests/core/test_registry_user_layer.py index bb5b3a22..ec962618 100644 --- a/tests/core/test_registry_user_layer.py +++ b/tests/core/test_registry_user_layer.py @@ -227,6 +227,8 @@ class TestLLMCreatorDispatchUsesUpstreamModelId: "supports_structured_output": False, "attachments": [], "context_window": 8192, + "api_flavor": "responses", + "reasoning_effort": "high", }, ) @@ -262,6 +264,8 @@ class TestLLMCreatorDispatchUsesUpstreamModelId: assert caps.supports_tools is False assert caps.supports_structured_output is False assert caps.supported_attachment_types == [] + assert caps.api_flavor == "responses" + assert caps.reasoning_effort == "high" def test_byom_image_alias_expands_to_mime_types(self, pg_conn): """A BYOM stored with ``attachments: ["image"]`` (the alias the diff --git a/tests/llm/test_openai_responses.py b/tests/llm/test_openai_responses.py index 127edf49..2a5eb408 100644 --- a/tests/llm/test_openai_responses.py +++ b/tests/llm/test_openai_responses.py @@ -50,6 +50,16 @@ def _responses_caps(reasoning_effort=None): ) +def _bare_agent(): + from application.agents.base import BaseAgent + + class _Agent(BaseAgent): + def _gen_inner(self, query, log_context): + yield from () + + return _Agent.__new__(_Agent) + + # ── api_flavor gating ──────────────────────────────────────────────────────── @@ -132,6 +142,154 @@ def test_to_responses_input_reinjects_reasoning(monkeypatch): assert items[1]["call_id"] == "call_1" +@pytest.mark.unit +def test_chat_flavor_drops_responses_only_message_fields(monkeypatch): + llm = _make_llm( + monkeypatch, + ModelCapabilities(api_flavor="chat_completions"), + ) + cleaned = llm._clean_messages_openai([{ + "role": "assistant", + "content": "answer", + "responses_reasoning_items": [{"type": "reasoning", "id": "rs_1"}], + }]) + assert cleaned == [{"role": "assistant", "content": "answer"}] + + +@pytest.mark.unit +def test_build_messages_gates_reasoning_and_keeps_tool_chronology(monkeypatch): + agent = _bare_agent() + agent.compressed_summary = None + agent.model_id = "model" + agent.model_user_id = None + agent.user = "u" + agent.multimodal_content = None + agent.llm = _ns( + _uses_responses_api=lambda: True, + responses_chain_key=lambda: "chain-current", + ) + agent.chat_history = [{ + "prompt": "old question", + "response": "final answer", + "tool_calls": [{ + "call_id": "call_1", + "action_name": "search", + "arguments": {"q": "x"}, + "result": "found", + }], + "metadata": { + "responses_state": { + "chain_key": "chain-other", + "reasoning_items": [{"type": "reasoning", "id": "wrong"}], + } + }, + }] + + messages = agent._build_messages("system", "new question") + assert [message["role"] for message in messages] == [ + "system", "user", "assistant", "tool", "assistant", "user" + ] + assert messages[2]["tool_calls"][0]["id"] == "call_1" + assert messages[4]["content"] == "final answer" + assert "responses_reasoning_items" not in messages[4] + + +@pytest.mark.unit +def test_build_messages_rewrites_duplicate_tool_call_ids(monkeypatch): + agent = _bare_agent() + agent.compressed_summary = None + agent.model_id = "model" + agent.model_user_id = None + agent.user = "u" + agent.multimodal_content = None + agent.llm = _ns( + _uses_responses_api=lambda: True, + responses_chain_key=lambda: "chain-current", + ) + agent.chat_history = [{ + "prompt": "old question", + "response": "final answer", + "tool_calls": [ + { + "call_id": "functions.search:0", + "action_name": "search", + "arguments": {"q": "first"}, + "result": "first result", + }, + { + "call_id": "functions.search:0", + "action_name": "search", + "arguments": {"q": "second"}, + "result": "second result", + }, + ], + }] + + messages = agent._build_messages("system", "new question") + assistant_calls = messages[2]["tool_calls"] + tool_results = [message for message in messages if message["role"] == "tool"] + + replay_ids = [call["id"] for call in assistant_calls] + assert replay_ids[0] == "functions.search:0" + assert len(set(replay_ids)) == 2 + assert [message["tool_call_id"] for message in tool_results] == replay_ids + assert [message["content"] for message in tool_results] == [ + "first result", + "second result", + ] + + +@pytest.mark.unit +def test_export_import_replays_encrypted_reasoning_on_fresh_instance(monkeypatch): + first = _make_llm(monkeypatch, _responses_caps()) + reasoning = { + "type": "reasoning", + "id": "rs_1", + "encrypted_content": "ciphertext", + "summary": [], + } + first._last_response_id = "resp_1" + first._last_reasoning_items = [reasoning] + first._reasoning_for_calls = {"call_1": [reasoning]} + + second = _make_llm(monkeypatch, _responses_caps()) + assert second.import_responses_state(first.export_responses_state()) is True + items = second._to_responses_input([{ + "role": "assistant", + "content": None, + "tool_calls": [{ + "id": "call_1", + "type": "function", + "function": {"name": "run", "arguments": "{}"}, + }], + }]) + assert items[0] == reasoning + assert items[1]["type"] == "function_call" + + +@pytest.mark.unit +def test_import_rejects_state_for_different_target(monkeypatch): + first = _make_llm(monkeypatch, _responses_caps()) + state = first.export_responses_state() + second = _make_llm(monkeypatch, _responses_caps()) + second._effective_base_url = "https://different.example/v1" + assert second.import_responses_state(state) is False + + +@pytest.mark.unit +def test_start_responses_turn_drops_prior_turn_call_map(monkeypatch): + llm = _make_llm(monkeypatch, _responses_caps()) + llm._reasoning_for_calls = {"old_call": [{"id": "old_reasoning"}]} + llm._last_reasoning_items = [{"id": "old_reasoning"}] + llm._imported_response_id = "resp_old" + + llm.start_responses_turn() + + assert llm._reasoning_for_calls == {} + assert llm._last_reasoning_items == [] + assert llm._imported_response_id is None + + @pytest.mark.unit def test_to_responses_input_multimodal_image(monkeypatch): llm = _make_llm(monkeypatch, _responses_caps()) @@ -227,6 +385,36 @@ def test_build_responses_params_stateless(monkeypatch): assert "previous_response_id" not in params +@pytest.mark.unit +def test_chat_path_drops_tool_controls_when_tools_are_unavailable(monkeypatch): + caps = ModelCapabilities( + supports_tools=False, + api_flavor="chat_completions", + ) + llm = _make_llm(monkeypatch, caps) + llm.client.chat.completions.create = MagicMock( + return_value=_ns(choices=[_ns(message=_ns(content="ok"))]) + ) + + result = llm._raw_gen( + llm, + "model", + [{"role": "user", "content": "hi"}], + tools=[{ + "type": "function", + "function": {"name": "search", "parameters": {}}, + }], + tool_choice="required", + parallel_tool_calls=True, + ) + + assert result == "ok" + request = llm.client.chat.completions.create.call_args.kwargs + assert "tools" not in request + assert "tool_choice" not in request + assert "parallel_tool_calls" not in request + + @pytest.mark.unit def test_build_responses_params_store_with_previous_id(monkeypatch): llm = _make_llm(monkeypatch, _responses_caps(), store_responses=True) @@ -241,6 +429,50 @@ def test_build_responses_params_store_with_previous_id(monkeypatch): assert params["include"] == ["reasoning.encrypted_content"] +@pytest.mark.unit +def test_build_responses_params_maps_tool_controls(monkeypatch): + llm = _make_llm(monkeypatch, _responses_caps()) + params = llm._build_responses_params( + "gpt-5.5", + [], + tools=[{"type": "function", "function": {"name": "search"}}], + response_format=None, + previous_response_id=None, + stream=False, + kwargs={ + "tool_choice": { + "type": "function", + "function": {"name": "search"}, + }, + "parallel_tool_calls": False, + }, + ) + assert params["tool_choice"] == {"type": "function", "name": "search"} + assert params["parallel_tool_calls"] is False + + +@pytest.mark.unit +def test_record_responses_metadata_captures_usage_details(monkeypatch): + llm = _make_llm(monkeypatch, _responses_caps()) + llm._record_responses_metadata(_ns( + id="resp_1", + usage=_ns( + input_tokens=10, + output_tokens=7, + total_tokens=17, + input_tokens_details=_ns(cached_tokens=4), + output_tokens_details=_ns(reasoning_tokens=3), + ), + )) + assert llm._last_usage == { + "prompt_tokens": 10, + "completion_tokens": 7, + "total_tokens": 17, + "prompt_tokens_details": {"cached_tokens": 4}, + "completion_tokens_details": {"reasoning_tokens": 3}, + } + + # ── streaming normalization into the existing handler contract ─────────────── @@ -343,6 +575,67 @@ def test_responses_gen_stream_error_event_raises(monkeypatch): list(llm._responses_gen_stream("gpt-5.5", [{"role": "user", "content": "hi"}], tools=None)) +@pytest.mark.unit +def test_responses_gen_stream_incomplete_returns_partial_length(monkeypatch): + llm = _make_llm(monkeypatch, _responses_caps()) + events = [ + _ns(type="response.output_text.delta", delta="partial"), + _ns( + type="response.incomplete", + response=_ns( + id="resp_incomplete", + incomplete_details=_ns(reason="max_output_tokens"), + ), + ), + ] + llm.client.responses.create = MagicMock(return_value=events) + + out = list( + llm._responses_gen_stream( + "gpt-5.5", [{"role": "user", "content": "hi"}], tools=None + ) + ) + from application.llm.handlers.openai import OpenAILLMHandler + + assert out[0] == "partial" + assert OpenAILLMHandler().parse_response(out[-1]).finish_reason == "length" + assert llm._last_response_id == "resp_incomplete" + + +@pytest.mark.unit +def test_responses_gen_stream_surfaces_refusal(monkeypatch): + llm = _make_llm(monkeypatch, _responses_caps()) + events = [ + _ns(type="response.refusal.delta", delta="I cannot"), + _ns(type="response.refusal.delta", delta=" help with that."), + _ns(type="response.completed", response=_ns(id="resp_refusal")), + ] + llm.client.responses.create = MagicMock(return_value=events) + + assert list( + llm._responses_gen_stream( + "gpt-5.5", [{"role": "user", "content": "hi"}], tools=None + ) + ) == ["I cannot", " help with that."] + assert llm._last_response_id == "resp_refusal" + + +@pytest.mark.unit +def test_responses_gen_stream_surfaces_done_only_refusal(monkeypatch): + llm = _make_llm(monkeypatch, _responses_caps()) + events = [ + _ns(type="response.refusal.done", refusal="I cannot help with that."), + _ns(type="response.completed", response=_ns(id="resp_refusal")), + ] + llm.client.responses.create = MagicMock(return_value=events) + + assert list( + llm._responses_gen_stream( + "gpt-5.5", [{"role": "user", "content": "hi"}], tools=None + ) + ) == ["I cannot help with that."] + + @pytest.mark.unit def test_responses_gen_nonstream_tools(monkeypatch): from application.llm.handlers.openai import OpenAILLMHandler @@ -376,6 +669,185 @@ def test_responses_gen_nonstream_text(monkeypatch): assert result == "Hi there" +@pytest.mark.unit +def test_responses_gen_nonstream_surfaces_refusal(monkeypatch): + llm = _make_llm(monkeypatch, _responses_caps()) + response = _ns( + id="resp_refusal", + status="completed", + output=[ + _ns( + type="message", + content=[_ns(type="refusal", refusal="I cannot help with that.")], + ) + ], + ) + llm.client.responses.create = MagicMock(return_value=response) + + result = llm._responses_gen( + "gpt-5.5", [{"role": "user", "content": "hi"}], tools=None + ) + assert result == "I cannot help with that." + assert llm._last_response_id == "resp_refusal" + + +@pytest.mark.unit +def test_responses_gen_nonstream_incomplete_returns_partial_length(monkeypatch): + llm = _make_llm(monkeypatch, _responses_caps()) + response = _ns( + id="resp_incomplete", + status="incomplete", + incomplete_details=_ns(reason="max_output_tokens"), + output=[ + _ns(type="message", content=[_ns(type="output_text", text="partial")]) + ], + ) + llm.client.responses.create = MagicMock(return_value=response) + + result = llm._responses_gen( + "gpt-5.5", [{"role": "user", "content": "hi"}], tools=None + ) + assert result == "partial" + assert llm._last_finish_reason == "length" + assert llm._last_response_id == "resp_incomplete" + + +@pytest.mark.unit +def test_public_gen_keeps_plain_string_contract_for_incomplete_text(monkeypatch): + llm = _make_llm(monkeypatch, _responses_caps()) + response = _ns( + id="resp_incomplete", + status="incomplete", + incomplete_details=_ns(reason="max_output_tokens"), + output=[ + _ns(type="message", content=[ + _ns(type="output_text", text="partial title") + ]) + ], + ) + llm.client.responses.create = MagicMock(return_value=response) + monkeypatch.setattr("application.cache.get_redis_instance", lambda: None) + + result = llm.gen( + model="gpt-5.5", + messages=[{"role": "user", "content": "make a title"}], + ) + + assert result == "partial title" + assert isinstance(result, str) + assert result.strip() == "partial title" + assert llm._last_finish_reason == "length" + + +@pytest.mark.unit +def test_stream_error_event_preserves_upstream_message(monkeypatch): + llm = _make_llm(monkeypatch, _responses_caps()) + llm.client.responses.create = MagicMock( + return_value=[_ns(type="error", message="real upstream failure")] + ) + with pytest.raises(RuntimeError, match="real upstream failure"): + list( + llm._responses_gen_stream( + "gpt-5.5", [{"role": "user", "content": "hi"}], tools=None + ) + ) + + +@pytest.mark.unit +def test_previous_response_id_requires_immediately_preceding_matching_chain(): + agent = _bare_agent() + agent.llm = _ns(responses_chain_key=lambda: "chain-current") + + agent.chat_history = [{ + "metadata": { + "response_id": "resp_ok", + "response_chain_key": "chain-current", + } + }] + assert agent._previous_response_id() == "resp_ok" + + agent.chat_history.append({"metadata": {}}) + assert agent._previous_response_id() is None + + agent.chat_history[-1] = { + "metadata": { + "response_id": "resp_other", + "response_chain_key": "chain-other", + } + } + assert agent._previous_response_id() is None + + +@pytest.mark.unit +def test_responses_chain_key_scopes_model_endpoint_and_credential(monkeypatch): + llm = _make_llm(monkeypatch, _responses_caps()) + llm._canonical_model_id = "model-a" + initial = llm.responses_chain_key() + + llm._canonical_model_id = "model-b" + different_model = llm.responses_chain_key() + llm._canonical_model_id = "model-a" + llm._effective_base_url = "https://other.example/v1" + different_endpoint = llm.responses_chain_key() + llm._effective_base_url = "https://api.openai.com/v1" + llm.api_key = "different-key" + different_credential = llm.responses_chain_key() + + llm.api_key = "k" + monkeypatch.setattr( + "application.llm.openai.settings.OPENAI_RESPONSES_STORE", True + ) + different_store_mode = llm.responses_chain_key() + + assert len({ + initial, + different_model, + different_endpoint, + different_credential, + different_store_mode, + }) == 5 + assert "different-key" not in different_credential + + +@pytest.mark.unit +def test_responses_metadata_persists_chain_key(monkeypatch): + monkeypatch.setattr( + "application.agents.base.settings.OPENAI_RESPONSES_STORE", True + ) + agent = _bare_agent() + agent.llm = _ns( + _last_response_id="resp_1", + responses_chain_key=lambda: "chain-current", + ) + + assert list(agent._emit_responses_metadata()) == [{ + "metadata": { + "response_id": "resp_1", + "response_chain_key": "chain-current", + } + }] + + +@pytest.mark.unit +def test_store_false_metadata_omits_unstored_response_id(monkeypatch): + monkeypatch.setattr( + "application.agents.base.settings.OPENAI_RESPONSES_STORE", False + ) + agent = _bare_agent() + agent.llm = _ns( + _last_response_id="resp_unstored", + _last_usage=None, + _uses_responses_api=lambda: True, + responses_chain_key=lambda: "chain-current", + export_responses_state=lambda: {"reasoning_items": [{"id": "rs_1"}]}, + ) + + metadata = list(agent._emit_responses_metadata())[0]["metadata"] + assert "response_id" not in metadata + assert "response_chain_key" not in metadata + assert metadata["responses_state"]["reasoning_items"][0]["id"] == "rs_1" + + # ── capability plumbing / yaml ─────────────────────────────────────────────── diff --git a/tests/storage/db/repositories/test_pending_tool_state.py b/tests/storage/db/repositories/test_pending_tool_state.py index 4ee250aa..05f65fd7 100644 --- a/tests/storage/db/repositories/test_pending_tool_state.py +++ b/tests/storage/db/repositories/test_pending_tool_state.py @@ -109,7 +109,7 @@ class TestMarkResuming: flipped = repo.mark_resuming(conv["id"], "user-1") assert flipped is True - loaded = repo.load_state(conv["id"], "user-1") + loaded = repo.load_state_any(conv["id"], "user-1") assert loaded["status"] == "resuming" assert loaded["resumed_at"] is not None @@ -181,7 +181,7 @@ class TestRevertStaleResuming: reverted = repo.revert_stale_resuming(grace_seconds=600) assert reverted == 0 - loaded = repo.load_state(conv["id"], "user-1") + loaded = repo.load_state_any(conv["id"], "user-1") assert loaded["status"] == "resuming" def test_leaves_pending_alone(self, pg_conn): @@ -201,9 +201,17 @@ class TestRevertStaleResuming: # so the user gets a fresh window to retry. conv = _conv(pg_conn) repo = _repo(pg_conn) - repo.save_state(conv["id"], "user-1", **_sample_state(), ttl_seconds=0) + repo.save_state(conv["id"], "user-1", **_sample_state()) repo.mark_resuming(conv["id"], "user-1") self._backdate_resumed(pg_conn, conv["id"], "user-1", 660) + pg_conn.execute( + text( + "UPDATE pending_tool_state " + "SET expires_at = clock_timestamp() - interval '1 second' " + "WHERE conversation_id = CAST(:conv_id AS uuid)" + ), + {"conv_id": conv["id"]}, + ) repo.revert_stale_resuming(grace_seconds=600) # Same tick: the original expires_at was set to "now" (ttl=0) diff --git a/tests/storage/db/repositories/test_user_custom_models.py b/tests/storage/db/repositories/test_user_custom_models.py index c8a5847b..7cfeac51 100644 --- a/tests/storage/db/repositories/test_user_custom_models.py +++ b/tests/storage/db/repositories/test_user_custom_models.py @@ -51,6 +51,20 @@ class TestCreate: "context_window": 200_000, } + def test_responses_capabilities_are_persisted(self, pg_conn): + repo = _repo(pg_conn) + row = _make( + repo, + capabilities={ + "api_flavor": "responses", + "reasoning_effort": "high", + }, + ) + assert row["capabilities"] == { + "api_flavor": "responses", + "reasoning_effort": "high", + } + class TestGet: def test_get_by_id_returns_row(self, pg_conn): diff --git a/tests/test_cache.py b/tests/test_cache.py index 74b4ae9c..16118960 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -127,6 +127,84 @@ def test_stream_cache_miss(mock_make_redis): mock_redis_instance.set.assert_called_once() +@pytest.mark.unit +@patch("application.cache.get_redis_instance") +def test_stream_cache_preserves_json_chunk_types(mock_make_redis): + mock_redis_instance = MagicMock() + mock_make_redis.return_value = mock_redis_instance + mock_redis_instance.get.return_value = json.dumps({ + "version": 1, + "chunks": ["text", {"type": "thought", "thought": "reasoning"}], + }).encode("utf-8") + + @stream_cache + def mock_function(self, model, messages, stream, tools): + yield "new_chunk" + + result = list(mock_function( + None, + "model", + [{"role": "user", "content": "test"}], + stream=True, + tools=None, + )) + + assert result == ["text", {"type": "thought", "thought": "reasoning"}] + mock_redis_instance.set.assert_not_called() + + +@pytest.mark.unit +@patch("application.cache.get_redis_instance") +def test_stream_cache_does_not_stringify_protocol_objects(mock_make_redis): + mock_redis_instance = MagicMock() + mock_make_redis.return_value = mock_redis_instance + mock_redis_instance.get.return_value = None + terminal_chunk = MagicMock(name="responses_terminal_choice") + + @stream_cache + def mock_function(self, model, messages, stream, tools): + yield "partial" + yield terminal_chunk + + result = list(mock_function( + None, + "model", + [{"role": "user", "content": "test"}], + stream=True, + tools=None, + )) + + assert result[0] == "partial" + assert result[1] is terminal_chunk + mock_redis_instance.set.assert_not_called() + + +@pytest.mark.unit +@patch("application.cache.get_redis_instance") +def test_stream_cache_rejects_legacy_protocol_object_repr(mock_make_redis): + mock_redis_instance = MagicMock() + mock_make_redis.return_value = mock_redis_instance + mock_redis_instance.get.return_value = json.dumps([ + "partial", + "", + ]).encode("utf-8") + + @stream_cache + def mock_function(self, model, messages, stream, tools): + yield "fresh" + + result = list(mock_function( + None, + "model", + [{"role": "user", "content": "test"}], + stream=True, + tools=None, + )) + + assert result == ["fresh"] + mock_redis_instance.delete.assert_called_once() + + # ── get_redis_instance ────────────────────────────────────────────────────── diff --git a/tests/test_continuation.py b/tests/test_continuation.py index d068e545..842d15db 100644 --- a/tests/test_continuation.py +++ b/tests/test_continuation.py @@ -838,12 +838,10 @@ class TestValidateRequest: @pytest.mark.unit class TestResumeMarkResuming: - """Resumed runs must mark state ``resuming`` instead of deleting it - eagerly; the row stays in PG so a crashed resume can be retried.""" + """Resumed runs atomically claim state instead of deleting it eagerly.""" - def test_resume_calls_mark_resuming_not_delete(self, monkeypatch): - """``resume_from_tool_actions`` flips the row to 'resuming' and - does not delete it before the run finishes.""" + def test_resume_claims_state_once_and_does_not_delete(self, monkeypatch): + """``resume_from_tool_actions`` consumes one atomic claim.""" from application.api.answer.services import ( continuation_service as cont_mod, ) @@ -852,7 +850,7 @@ class TestResumeMarkResuming: from application.llm.handlers import handler_creator as handler_mod cont_service = MagicMock() - cont_service.load_state.return_value = { + cont_service.claim_state.return_value = { "messages": [], "pending_tool_calls": [], "tools_dict": {}, @@ -871,8 +869,6 @@ class TestResumeMarkResuming: }, "client_tools": None, } - cont_service.mark_resuming.return_value = True - monkeypatch.setattr( cont_mod, "ContinuationService", lambda: cont_service ) @@ -908,7 +904,7 @@ class TestResumeMarkResuming: conversation_id="00000000-0000-0000-0000-000000000001", ) - cont_service.mark_resuming.assert_called_once_with( + cont_service.claim_state.assert_called_once_with( "00000000-0000-0000-0000-000000000001", "alice" ) cont_service.delete_state.assert_not_called() @@ -929,7 +925,7 @@ class TestResumeMarkResuming: reserved_id = "22222222-2222-2222-2222-222222222222" cont_service = MagicMock() - cont_service.load_state.return_value = { + cont_service.claim_state.return_value = { "messages": [], "pending_tool_calls": [], "tools_dict": {}, @@ -949,7 +945,6 @@ class TestResumeMarkResuming: }, "client_tools": None, } - cont_service.mark_resuming.return_value = True monkeypatch.setattr(cont_mod, "ContinuationService", lambda: cont_service) monkeypatch.setattr( llm_creator_mod.LLMCreator, "create_llm", lambda *a, **kw: MagicMock(), @@ -1004,7 +999,7 @@ class TestResumeMarkResuming: from application.llm.handlers import handler_creator as handler_mod cont_service = MagicMock() - cont_service.load_state.return_value = { + cont_service.claim_state.return_value = { "messages": [], "pending_tool_calls": [], "tools_dict": {}, @@ -1023,7 +1018,6 @@ class TestResumeMarkResuming: }, "client_tools": None, } - cont_service.mark_resuming.return_value = True monkeypatch.setattr(cont_mod, "ContinuationService", lambda: cont_service) monkeypatch.setattr( llm_creator_mod.LLMCreator, "create_llm", lambda *a, **kw: MagicMock(), @@ -1066,8 +1060,7 @@ class TestResumeMarkResuming: fake_repo.find_by_key.assert_called_once_with("agent-key-1") # The lookup + claim now key on the owner id, not None. - cont_service.load_state.assert_called_once_with(conv_id, "owner-1") - cont_service.mark_resuming.assert_called_once_with(conv_id, "owner-1") + cont_service.claim_state.assert_called_once_with(conv_id, "owner-1") assert sp.initial_user_id == "owner-1" assert sp.decoded_token == {"sub": "owner-1"} @@ -1120,7 +1113,7 @@ class TestContinuationServiceMarkResuming: assert flipped is True with pg_engine.connect() as conn: - row = PendingToolStateRepository(conn).load_state( + row = PendingToolStateRepository(conn).load_state_any( conv["id"], "alice" ) assert row["status"] == "resuming" diff --git a/tests/test_v1_translator.py b/tests/test_v1_translator.py index ae3156fe..585fd5d5 100644 --- a/tests/test_v1_translator.py +++ b/tests/test_v1_translator.py @@ -9,6 +9,7 @@ import json import pytest from application.api.v1.translator import ( + StreamTranslationState, _get_client_tool_name, _split_leaked_reasoning, _strip_repr_quotes, @@ -172,13 +173,13 @@ class TestExtractSystemPrompt: messages = [{"role": "user", "content": "Hello"}] assert extract_system_prompt(messages) is None - def test_returns_first_of_multiple_system_messages(self): + def test_combines_multiple_instruction_messages_in_order(self): messages = [ {"role": "system", "content": "First"}, - {"role": "system", "content": "Second"}, + {"role": "developer", "content": "Second"}, {"role": "user", "content": "Hello"}, ] - assert extract_system_prompt(messages) == "First" + assert extract_system_prompt(messages) == "First\n\nSecond" def test_empty_content_returns_empty_string(self): messages = [ @@ -524,6 +525,18 @@ class TestTranslateResponse: ) assert "docsgpt" not in resp + def test_finish_reason_override(self): + resp = translate_response( + conversation_id="c1", + answer="Partial output", + sources=None, + tool_calls=None, + thought="", + model_name="agent", + finish_reason_override="length", + ) + assert resp["choices"][0]["finish_reason"] == "length" + # --------------------------------------------------------------------------- # translate_stream_event @@ -573,6 +586,14 @@ class TestTranslateStreamEvent: # Second chunk: [DONE] assert chunks[1].strip() == "data: [DONE]" + def test_end_event_preserves_length_finish_reason(self): + chunks = translate_stream_event( + {"type": "end", "finish_reason": "length"}, + "chatcmpl-1", "agent", + ) + parsed = json.loads(chunks[0].replace("data: ", "").strip()) + assert parsed["choices"][0]["finish_reason"] == "length" + def test_tool_call_client_execution(self): chunks = translate_stream_event( { @@ -645,6 +666,52 @@ class TestTranslateStreamEvent: ext = json.loads(chunks[1].replace("data: ", "").strip()) assert ext["docsgpt"]["type"] == "tool_calls_pending" + def test_parallel_tool_indices_and_single_terminal_reason(self): + state = StreamTranslationState() + chunks = [] + for call_id in ("c1", "c2"): + chunks += translate_stream_event( + { + "type": "tool_call", + "data": { + "call_id": call_id, + "action_name": "run", + "arguments": {}, + "status": "requires_client_execution", + }, + }, + "chatcmpl-1", + "agent", + state=state, + ) + chunks += translate_stream_event( + {"type": "tool_calls_pending", "data": {"pending_tool_calls": []}}, + "chatcmpl-1", + "agent", + state=state, + ) + chunks += translate_stream_event( + {"type": "end"}, "chatcmpl-1", "agent", state=state + ) + decoded = [ + json.loads(chunk.removeprefix("data: ").strip()) + for chunk in chunks + if "[DONE]" not in chunk + ] + indices = [ + item["choices"][0]["delta"]["tool_calls"][0]["index"] + for item in decoded + if item["choices"][0]["delta"].get("tool_calls") + ] + finishes = [ + item["choices"][0]["finish_reason"] + for item in decoded + if item["choices"][0]["finish_reason"] + ] + assert indices == [0, 1] + assert finishes == ["tool_calls"] + assert sum("[DONE]" in chunk for chunk in chunks) == 1 + def test_id_event(self): chunks = translate_stream_event( {"type": "id", "id": "conv-123"}, @@ -931,6 +998,21 @@ class TestTranslateRequestStructuredOutputs: result = translate_request(data, "key") assert result["llm_params"] == {"temperature": 0.2, "top_p": 0.9, "seed": 7} + def test_tool_controls_are_forwarded(self): + choice = {"type": "function", "function": {"name": "read_file"}} + data = { + "messages": [{"role": "user", "content": "Hi"}], + "tools": [{ + "type": "function", + "function": {"name": "read_file", "parameters": {}}, + }], + "tool_choice": choice, + "parallel_tool_calls": False, + } + result = translate_request(data, "key") + assert result["llm_params"]["tool_choice"] == choice + assert result["llm_params"]["parallel_tool_calls"] is False + def test_max_tokens_alias_dropped_when_canonical_present(self): data = { "messages": [{"role": "user", "content": "Hi"}],