feat: improved compatability layer and responses api handing

This commit is contained in:
Alex committed 2026-07-13 09:37:58 +01:00
1 parent 3e978759de
commit 0adf88abed
45 files changed
+2566 -337

No files matched your search

+1
View File
@@ -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
+168 -47
View File
@@ -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.
+73 -51
View File
@@ -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
@@ -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
@@ -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.
@@ -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,
@@ -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,
+66 -13
View File
@@ -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)
+26
View File
@@ -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",
+215 -40
View File
@@ -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)
+99
View File
@@ -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)
+76 -19
View File
@@ -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
+38 -9
View File
@@ -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)
+4
View File
@@ -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(
+2
View File
@@ -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,
}
+1
View File
@@ -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: <string> # default null; none|minimal|low|medium|high|xhigh (subset is model-dependent)
api_flavor: <string> # chat_completions (default) or responses
models: # required
- id: <string, required> # unique registry key; persisted in agent records
+9
View File
@@ -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 = [
+238 -16
View File
@@ -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()
@@ -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,
)
@@ -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()]
@@ -34,6 +34,8 @@ _ALLOWED_CAPABILITY_KEYS = frozenset(
"supports_streaming",
"attachments",
"context_window",
"api_flavor",
"reasoning_effort",
}
)
+17
View File
@@ -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)
@@ -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
+3 -2
View File
@@ -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.
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.
@@ -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<CustomModelCapabilities, 'api_flavor'>;
},
token: string | null,
): Promise<CustomModelTestResult> => {
@@ -122,14 +124,17 @@ const customModelsService = {
base_url?: string;
api_key?: string;
upstream_model_id?: string;
capabilities?: Pick<CustomModelCapabilities, 'api_flavor'>;
} = {},
): Promise<CustomModelTestResult> => {
// Send only non-empty overrides; server falls back to stored values.
const requestBody: Record<string, string> = {};
const requestBody: Record<string, unknown> = {};
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,
+9
View File
@@ -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",
+110 -7
View File
@@ -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 */}
<div className="flex flex-col gap-2">
<Label>{t('settings.customModels.capabilities.title')}</Label>
<div className="grid gap-3 sm:grid-cols-2">
<div className="flex flex-col gap-1.5">
<Label
htmlFor="cm-api-flavor"
className="text-muted-foreground text-xs"
>
{t('settings.customModels.capabilities.apiFlavor')}
</Label>
<Select
value={formData.api_flavor}
onValueChange={(value) =>
handleChange('api_flavor', value as ModelApiFlavor)
}
>
<SelectTrigger id="cm-api-flavor" className="w-full">
<SelectValue />
</SelectTrigger>
<SelectContent>
<SelectItem value="chat_completions">
{t(
'settings.customModels.capabilities.apiFlavors.chatCompletions',
)}
</SelectItem>
<SelectItem value="responses">
{t(
'settings.customModels.capabilities.apiFlavors.responses',
)}
</SelectItem>
</SelectContent>
</Select>
</div>
<div className="flex flex-col gap-1.5">
<Label
htmlFor="cm-reasoning-effort"
className="text-muted-foreground text-xs"
>
{t('settings.customModels.capabilities.reasoningEffort')}
</Label>
<Select
value={formData.reasoning_effort}
onValueChange={(value) =>
handleChange(
'reasoning_effort',
value as ReasoningEffort | 'default',
)
}
>
<SelectTrigger id="cm-reasoning-effort" className="w-full">
<SelectValue />
</SelectTrigger>
<SelectContent>
<SelectItem value="default">
{t(
'settings.customModels.capabilities.reasoningEfforts.default',
)}
</SelectItem>
{(
[
'none',
'minimal',
'low',
'medium',
'high',
'xhigh',
] as ReasoningEffort[]
).map((effort) => (
<SelectItem key={effort} value={effort}>
{effort}
</SelectItem>
))}
</SelectContent>
</Select>
</div>
</div>
<div className="flex flex-wrap gap-2">
<CapabilityChip
label={t('settings.customModels.capabilities.chips.tools')}
+10
View File
@@ -1,4 +1,12 @@
export type ModelSource = 'builtin' | 'user';
export type ModelApiFlavor = 'chat_completions' | 'responses';
export type ReasoningEffort =
| 'none'
| 'minimal'
| 'low'
| 'medium'
| 'high'
| 'xhigh';
export interface AvailableModel {
id: string;
@@ -33,6 +41,8 @@ export interface CustomModelCapabilities {
supports_structured_output: boolean;
attachments: string[];
context_window: number;
api_flavor?: ModelApiFlavor;
reasoning_effort?: ReasoningEffort;
}
export interface CustomModel {
@@ -4,6 +4,8 @@ from contextlib import contextmanager
from unittest.mock import patch
from uuid import uuid4
import pytest
from sqlalchemy import text
@contextmanager
@@ -161,6 +163,68 @@ class TestContinuationServiceSaveLoad:
)
assert got is None
def test_claim_state_is_atomic_and_duplicate_is_conflict(self, pg_conn):
from application.api.answer.services.continuation_service import (
ContinuationService,
ResumeInProgressError,
)
from application.storage.db.repositories.conversations import (
ConversationsRepository,
)
user = "u-claim"
conv = ConversationsRepository(pg_conn).create(user, name="c")
conv_id = str(conv["id"])
service = ContinuationService()
with _patch_db(pg_conn):
service.save_state(
conversation_id=conv_id,
user=user,
messages=[],
pending_tool_calls=[{"call_id": "call-1"}],
tools_dict={},
tool_schemas=[],
agent_config={"agent_id": "agent-a"},
)
claimed = service.claim_state(conv_id, user)
assert claimed is not None
assert claimed["status"] == "resuming"
with pytest.raises(ResumeInProgressError):
service.claim_state(conv_id, user)
def test_expired_state_is_neither_loaded_nor_claimed(self, pg_conn):
from application.api.answer.services.continuation_service import (
ContinuationService,
)
from application.storage.db.repositories.conversations import (
ConversationsRepository,
)
user = "u-expired"
conv = ConversationsRepository(pg_conn).create(user, name="c")
conv_id = str(conv["id"])
service = ContinuationService()
with _patch_db(pg_conn):
service.save_state(
conversation_id=conv_id,
user=user,
messages=[],
pending_tool_calls=[{"call_id": "call-1"}],
tools_dict={},
tool_schemas=[],
agent_config={},
)
pg_conn.execute(
text(
"UPDATE pending_tool_state "
"SET expires_at = clock_timestamp() - interval '1 second' "
"WHERE conversation_id = CAST(:conv AS uuid)"
),
{"conv": conv_id},
)
assert service.load_state(conv_id, user) is None
assert service.claim_state(conv_id, user) is None
def test_save_state_no_client_tools(self, pg_conn):
from application.api.answer.services.continuation_service import (
ContinuationService,
+16 -3
View File
@@ -485,7 +485,11 @@ class TestCleanupPendingToolState:
# Pending but already expired — should be deleted.
c2 = ConversationsRepository(pg_conn).create("u", "expired-pending")
repo.save_state(c2["id"], "u", **_sample(), ttl_seconds=0)
expired_state = _sample()
expired_state["agent_config"] = {
"reserved_message_id": "22222222-2222-2222-2222-222222222222"
}
repo.save_state(c2["id"], "u", **expired_state, ttl_seconds=0)
# Resuming within grace — should stay 'resuming'.
c3 = ConversationsRepository(pg_conn).create("u", "fresh-resuming")
@@ -518,16 +522,25 @@ class TestCleanupPendingToolState:
with patch(
"application.storage.db.engine.get_engine",
return_value=fake_engine,
):
), patch(
"application.api.answer.services.conversation_service."
"ConversationService.finalize_message",
) as finalize_expired:
result = cleanup_pending_tool_state.run()
assert result["reverted"] == 1
assert result["deleted"] == 1
finalize_expired.assert_called_once()
assert finalize_expired.call_args.args[0] == (
"22222222-2222-2222-2222-222222222222"
)
assert finalize_expired.call_args.kwargs["status"] == "failed"
# Final state assertions.
assert repo.load_state(c1["id"], "u")["status"] == "pending"
assert repo.load_state(c2["id"], "u") is None
assert repo.load_state(c3["id"], "u")["status"] == "resuming"
assert repo.load_state(c3["id"], "u") is None
assert repo.load_state_any(c3["id"], "u")["status"] == "resuming"
c4_row = repo.load_state(c4["id"], "u")
assert c4_row["status"] == "pending"
assert c4_row["resumed_at"] is None
@@ -227,6 +227,72 @@ class TestCreate:
resp = UserModelsCollectionResource().post()
assert resp.status_code == 201
def test_create_accepts_responses_capabilities(self, app, pg_conn):
from application.api.user.models.routes import (
UserModelsCollectionResource,
)
with patch("application.security.safe_url.socket.getaddrinfo") as gai:
gai.return_value = [(None, None, None, None, ("104.18.0.1", 0))]
with app.test_request_context(
"/api/user/models",
method="POST",
json={
"upstream_model_id": "gpt-custom",
"display_name": "Responses model",
"base_url": "https://example.openai.azure.com/openai/v1",
"api_key": "k",
"capabilities": {
"api_flavor": "responses",
"reasoning_effort": "high",
},
},
):
from flask import request
request.decoded_token = {"sub": "user-1"}
with _patch_db(pg_conn):
resp = UserModelsCollectionResource().post()
assert resp.status_code == 201
assert resp.get_json()["capabilities"] == {
"api_flavor": "responses",
"reasoning_effort": "high",
}
@pytest.mark.parametrize(
("capability", "value"),
[("api_flavor", "grpc"), ("reasoning_effort", "extreme")],
)
def test_create_rejects_invalid_responses_capabilities(
self, app, pg_conn, capability, value
):
from application.api.user.models.routes import (
UserModelsCollectionResource,
)
with patch("application.security.safe_url.socket.getaddrinfo") as gai:
gai.return_value = [(None, None, None, None, ("104.18.0.1", 0))]
with app.test_request_context(
"/api/user/models",
method="POST",
json={
"upstream_model_id": "gpt-custom",
"display_name": "Bad model",
"base_url": "https://example.openai.azure.com/openai/v1",
"api_key": "k",
"capabilities": {capability: value},
},
):
from flask import request
request.decoded_token = {"sub": "user-1"}
with _patch_db(pg_conn):
resp = UserModelsCollectionResource().post()
assert resp.status_code == 400
assert capability in resp.get_json()["error"]
def test_create_rejects_private_ip_dns(self, app, pg_conn):
from application.api.user.models.routes import (
UserModelsCollectionResource,
@@ -529,6 +595,44 @@ class TestPayloadConnectionTest:
assert call_args.kwargs["headers"]["Authorization"] == "Bearer sk-mistral-test"
assert call_args.kwargs["json"]["model"] == "mistral-large-latest"
def test_payload_test_uses_responses_protocol(self, app, pg_conn):
from application.api.user.models.routes import (
UserModelTestPayloadResource,
)
with patch("application.api.user.models.routes.pinned_post") as rp:
rp.return_value = MagicMock(
status_code=200,
headers={"Content-Type": "application/json"},
text='{"ok": true}',
)
with app.test_request_context(
"/api/user/models/test",
method="POST",
json={
"base_url": "https://example.openai.azure.com/openai/v1",
"api_key": "foundry-key",
"upstream_model_id": "gpt-custom",
"capabilities": {"api_flavor": "responses"},
},
):
from flask import request
request.decoded_token = {"sub": "user-1"}
with _patch_db(pg_conn):
resp = UserModelTestPayloadResource().post()
assert resp.status_code == 200
call_args = rp.call_args
assert call_args.args[0].endswith("/responses")
assert call_args.kwargs["json"] == {
"model": "gpt-custom",
"input": "hi",
"max_output_tokens": 16,
"stream": False,
"store": False,
}
def test_payload_test_unauthenticated_returns_401(self, app, pg_conn):
from application.api.user.models.routes import (
UserModelTestPayloadResource,
+217 -3
View File
@@ -181,6 +181,11 @@ class TestListModelsExtra:
@contextmanager
def _patch_v1_db(conn):
from application.storage.db.repositories.agents import AgentsRepository
if AgentsRepository(conn).find_by_key("x") is None:
AgentsRepository(conn).create("u-test", "Test Agent", "published", key="x")
@contextmanager
def _yield():
yield conn
@@ -286,6 +291,217 @@ class TestChatCompletionsHappyPath:
assert resp.status_code == 400
assert resp.get_json()["error"]["type"] == "invalid_request"
def test_invalid_api_key_returns_401_before_processing(self, pg_conn):
app = _build_app()
with _patch_v1_db(pg_conn):
with app.test_client() as c:
resp = c.post(
"/v1/chat/completions",
headers={"Authorization": "Bearer invalid"},
json={"messages": [{"role": "user", "content": "Hi"}]},
)
assert resp.status_code == 401
def test_model_field_is_accepted_and_ignored(self, pg_conn):
app = _build_app()
fake_processor = MagicMock()
fake_processor.decoded_token = {"sub": "u-test"}
fake_processor.conversation_id = None
fake_processor.agent_config = {}
fake_processor.agent_id = None
fake_processor.model_id = "m"
fake_processor.build_agent.return_value = MagicMock()
fake_helper = MagicMock()
fake_helper.check_usage.return_value = None
fake_helper.process_response_stream.return_value = {
"error": None,
"conversation_id": "conv-1",
"answer": "ok",
"sources": [],
"tool_calls": [],
"thought": "",
"extra": {},
}
with _patch_v1_db(pg_conn), patch(
"application.api.v1.routes.StreamProcessor",
return_value=fake_processor,
), patch(
"application.api.v1.routes._V1AnswerHelper",
return_value=fake_helper,
):
with app.test_client() as c:
resp = c.post(
"/v1/chat/completions",
headers={"Authorization": "Bearer x"},
json={
"model": "another-agent",
"messages": [{"role": "user", "content": "Hi"}],
},
)
assert resp.status_code == 200
def test_null_n_is_treated_as_default(self, pg_conn):
app = _build_app()
with _patch_v1_db(pg_conn), patch(
"application.api.v1.routes.translate_request",
side_effect=ValueError("reached translator"),
):
with app.test_client() as c:
resp = c.post(
"/v1/chat/completions",
headers={"Authorization": "Bearer x"},
json={
"n": None,
"messages": [{"role": "user", "content": "Hi"}],
},
)
assert resp.status_code == 400
assert resp.get_json()["error"]["message"] == "Failed to process request"
def test_non_object_stream_options_returns_400(self, pg_conn):
app = _build_app()
with _patch_v1_db(pg_conn):
with app.test_client() as c:
resp = c.post(
"/v1/chat/completions",
headers={"Authorization": "Bearer x"},
json={
"stream": True,
"stream_options": True,
"messages": [{"role": "user", "content": "Hi"}],
},
)
assert resp.status_code == 400
assert resp.get_json()["error"]["type"] == "invalid_request_error"
def test_conversation_id_must_belong_to_authenticated_agent(self, pg_conn):
from application.storage.db.repositories.agents import AgentsRepository
from application.storage.db.repositories.conversations import (
ConversationsRepository,
)
agents = AgentsRepository(pg_conn)
first = agents.create("same-owner", "First", "published", key="agent-a")
agents.create("same-owner", "Second", "published", key="agent-b")
conversation = ConversationsRepository(pg_conn).create(
"same-owner", "private", agent_id=str(first["id"])
)
app = _build_app()
with _patch_v1_db(pg_conn):
with app.test_client() as c:
resp = c.post(
"/v1/chat/completions",
headers={"Authorization": "Bearer agent-b"},
json={
"model": "ignored",
"conversation_id": str(conversation["id"]),
"messages": [{"role": "user", "content": "secret?"}],
},
)
assert resp.status_code == 400
assert resp.get_json()["error"]["code"] == "conversation_not_found"
def test_stale_session_mapping_is_deleted_and_replaced(self, pg_conn):
app = _build_app()
fake_processor = MagicMock()
fake_processor.decoded_token = {"sub": "u-test"}
fake_processor.conversation_id = None
fake_processor.agent_config = {}
fake_processor.agent_id = None
fake_processor.model_id = "m"
fake_processor.build_agent.return_value = MagicMock()
fake_helper = MagicMock()
fake_helper.check_usage.return_value = None
fake_helper.process_response_stream.return_value = {
"error": None,
"conversation_id": "new-conversation",
"answer": "ok",
"sources": [],
"tool_calls": [],
"thought": "",
"extra": {},
}
with _patch_v1_db(pg_conn), patch(
"application.api.v1.routes.load_conversation",
return_value="deleted-conversation",
), patch(
"application.api.v1.routes._conversation_belongs_to_agent",
return_value=False,
), patch(
"application.api.v1.routes.delete_conversation",
) as delete_mapping, patch(
"application.api.v1.routes.StreamProcessor",
return_value=fake_processor,
), patch(
"application.api.v1.routes._V1AnswerHelper",
return_value=fake_helper,
):
with app.test_client() as c:
resp = c.post(
"/v1/chat/completions",
headers={
"Authorization": "Bearer x",
"X-DocsGPT-Session-ID": "session",
},
json={"messages": [{"role": "user", "content": "Hi"}]},
)
assert resp.status_code == 200
delete_mapping.assert_called_once()
def test_duplicate_resume_returns_conflict(self, pg_conn):
from application.api.answer.services.continuation_service import (
ResumeInProgressError,
)
app = _build_app()
fake_processor = MagicMock()
fake_processor.decoded_token = {"sub": "u-test"}
with _patch_v1_db(pg_conn), patch(
"application.api.v1.routes.translate_request",
return_value={
"conversation_id": "conv-1",
"tool_actions": [{"call_id": "call-1", "result": "ok"}],
"messages": [],
},
), patch(
"application.api.v1.routes._conversation_belongs_to_agent",
return_value=True,
), patch(
"application.api.v1.routes.StreamProcessor",
return_value=fake_processor,
), patch(
"application.api.v1.routes.ContinuationService.claim_state",
side_effect=ResumeInProgressError("Resume already in progress"),
):
with app.test_client() as c:
resp = c.post(
"/v1/chat/completions",
headers={"Authorization": "Bearer x"},
json={"messages": [{"role": "tool", "content": "ok"}]},
)
assert resp.status_code == 409
assert resp.get_json()["error"]["code"] == "resume_in_progress"
def test_unsupported_n_and_logprobs_are_explicit(self, pg_conn):
app = _build_app()
with _patch_v1_db(pg_conn):
with app.test_client() as c:
n_response = c.post(
"/v1/chat/completions",
headers={"Authorization": "Bearer x"},
json={"n": 2, "messages": [{"role": "user", "content": "Hi"}]},
)
logprobs_response = c.post(
"/v1/chat/completions",
headers={"Authorization": "Bearer x"},
json={
"logprobs": True,
"messages": [{"role": "user", "content": "Hi"}],
},
)
assert n_response.status_code == 400
assert logprobs_response.status_code == 400
def test_translate_error_returns_400(self, pg_conn):
app = _build_app()
with _patch_v1_db(pg_conn), patch(
@@ -542,7 +758,7 @@ class TestChatCompletionsHappyPath:
translated_chunks: list = []
def _fake_translate_stream_event(
event_data, completion_id, model_name, strip_reasoning_leak=False
event_data, completion_id, model_name, strip_reasoning_leak=False, state=None
):
translated_chunks.append(event_data)
return ['data: x\n\n']
@@ -583,5 +799,3 @@ class TestChatCompletionsHappyPath:
assert "message_id" not in types_translated
assert "id" in types_translated
assert "answer" in types_translated
+74
View File
@@ -0,0 +1,74 @@
"""Tests for privacy-preserving v1 client session correlation."""
from unittest.mock import MagicMock
import pytest
from application.api.v1.session_store import (
delete_conversation,
identify_session,
load_conversation,
save_conversation,
)
@pytest.mark.unit
def test_session_key_is_prompt_and_agent_scoped():
headers = {"X-Session-ID": "raw-secret-session"}
main = {"messages": [{"role": "system", "content": "coding"}]}
title = {"messages": [{"role": "system", "content": "title"}]}
one = identify_session(headers, main, "agent-1")
assert one != identify_session(headers, title, "agent-1")
assert one != identify_session(headers, main, "agent-2")
assert "raw-secret-session" not in one.key
@pytest.mark.unit
def test_explicit_docsgpt_session_precedes_generic_header():
data = {
"docsgpt": {"session_id": "explicit"},
"messages": [{"role": "user", "content": "hi"}],
}
explicit = identify_session({"X-Session-ID": "generic"}, data, "agent")
expected = identify_session(
{"X-DocsGPT-Session-ID": "explicit"},
{"messages": data["messages"]},
"agent",
)
assert explicit == expected
@pytest.mark.unit
def test_openai_user_field_is_not_a_chat_session():
data = {
"user": "stable-end-user-id",
"messages": [{"role": "user", "content": "hi"}],
}
assert identify_session({}, data, "agent") is None
@pytest.mark.unit
def test_save_and_load_use_bounded_redis_ttl(monkeypatch):
redis = MagicMock()
redis.get.return_value = b"conv-1"
monkeypatch.setattr(
"application.api.v1.session_store.get_redis_instance", lambda: redis
)
monkeypatch.setattr(
"application.api.v1.session_store.settings.V1_SESSION_TTL_SECONDS", 123
)
session = identify_session({"X-Session-ID": "s"}, {"messages": []}, "a")
save_conversation(session, "conv-1")
redis.set.assert_called_once_with(session.key, "conv-1", ex=123)
assert load_conversation(session) == "conv-1"
@pytest.mark.unit
def test_delete_conversation_invalidates_stale_mapping(monkeypatch):
redis = MagicMock()
monkeypatch.setattr(
"application.api.v1.session_store.get_redis_instance", lambda: redis
)
session = identify_session({"X-Session-ID": "s"}, {"messages": []}, "a")
delete_conversation(session)
redis.delete.assert_called_once_with(session.key)
+22 -30
View File
@@ -1,25 +1,9 @@
"""Foreign-``conversation_id`` ownership gate on the ``/v1`` continuation path.
"""Owner-and-agent ``conversation_id`` gates on the ``/v1`` continuation path.
A ``/v1`` tool continuation persists its final answer into a client-supplied
``conversation_id``. Ownership IS enforced deep in the stack:
route → ``build_continuation_from_messages`` → ``build_agent("")``
→ ``initialize`` → ``_load_conversation_history``
→ ``get_conversation(conversation_id, owner)`` ← owner-scoped read
→ ``None`` for a foreign conversation
→ ``ValueError("Conversation not found or unauthorized")``
→ route's ``except ValueError`` → HTTP 400
…but the existing E2E suite mocks ``build_agent``, so this security gate is
never actually exercised. This test seeds a conversation owned by user A and an
agent owned by user B, then POSTs a continuation with B's api_key targeting A's
conversation. The ownership read runs against the **real** DB (not mocked): we
only short-circuit ``StreamProcessor.initialize`` to the
``_load_conversation_history`` step so the test doesn't have to stand up the
full agent/model/source/retriever machinery to reach the gate.
Asserted: the route returns 400 AND no assistant turn is appended to A's
conversation (the foreign write is rejected before persistence).
The route validates conversation metadata before initializing the stream
processor. These tests use the real database to verify that a foreign-owner or
same-owner/different-agent conversation cannot receive a continuation and that
the rejected request never appends an assistant turn.
"""
from __future__ import annotations
@@ -58,10 +42,12 @@ CLIENT_TOOL = {
}
def _seed_conversation(conn, user_id: str, name: str) -> 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()
+41 -29
View File
@@ -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
+1 -1
View File
@@ -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": {},
+16 -3
View File
@@ -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):
+31 -1
View File
@@ -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 ──────────────────────────────────────
+4
View File
@@ -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
+472
View File
@@ -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 ───────────────────────────────────────────────
@@ -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)
@@ -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):
+78
View File
@@ -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",
"<application.llm.openai._RespChoice object at 0x123>",
]).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 ──────────────────────────────────────────────────────
+9 -16
View File
@@ -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"
+85 -3
View File
@@ -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"}],