diff --git a/README.md b/README.md index 94b78e6..232ffff 100644 --- a/README.md +++ b/README.md @@ -103,6 +103,7 @@ account. The current catalog commonly includes: ## Features - Tool / function calling +- Structured outputs via JSON schema (`response_format`, or `format` on the Ollama routes) - Vision / image input - Thinking summaries (via think tags) - Configurable thinking effort diff --git a/chatmock/reasoning.py b/chatmock/reasoning.py index 37c276c..162692b 100644 --- a/chatmock/reasoning.py +++ b/chatmock/reasoning.py @@ -18,14 +18,20 @@ def build_reasoning_param( valid_efforts = allowed_efforts or DEFAULT_REASONING_EFFORTS valid_summaries = {"auto", "concise", "detailed", "none"} + # A caller's explicit effort is forwarded when upstream knows it at all; + # upstream validates per model. Filtering against the catalog here silently + # rewrote efforts the catalog omits but upstream honours (e.g. `none`). + explicit_effort: str | None = None if isinstance(overrides, dict): o_eff = str(overrides.get("effort", "")).strip().lower() o_sum = str(overrides.get("summary", "")).strip().lower() - if o_eff in valid_efforts and o_eff: - effort = o_eff + if o_eff in DEFAULT_REASONING_EFFORTS: + explicit_effort = o_eff if o_sum in valid_summaries and o_sum: summary = o_sum - if effort not in valid_efforts: + if explicit_effort is not None: + effort = explicit_effort + elif effort not in valid_efforts: effort = "medium" if summary not in valid_summaries: summary = "auto" diff --git a/chatmock/routes_ollama.py b/chatmock/routes_ollama.py index 70981ee..6048011 100644 --- a/chatmock/routes_ollama.py +++ b/chatmock/routes_ollama.py @@ -16,9 +16,18 @@ build_reasoning_param, extract_reasoning_from_model_name, ) -from .transform import convert_ollama_messages, normalize_ollama_tools +from .transform import ( + convert_ollama_format_to_text_format, + convert_ollama_messages, + normalize_ollama_tools, + ollama_format_requests_json, +) from .upstream import normalize_model_name, start_upstream_request -from .utils import convert_chat_messages_to_responses_input, convert_tools_chat_to_responses +from .utils import ( + convert_chat_messages_to_responses_input, + convert_tool_choice_chat_to_responses, + convert_tools_chat_to_responses, +) ollama_bp = Blueprint("ollama", __name__) @@ -183,21 +192,24 @@ def ollama_chat() -> Response: messages = convert_ollama_messages( raw_messages, payload.get("images") if isinstance(payload.get("images"), list) else None ) - if isinstance(messages, list): - sys_idx = next((i for i, m in enumerate(messages) if isinstance(m, dict) and m.get("role") == "system"), None) - if isinstance(sys_idx, int): - sys_msg = messages.pop(sys_idx) - content = sys_msg.get("content") if isinstance(sys_msg, dict) else "" - messages.insert(0, {"role": "user", "content": content}) stream_req = payload.get("stream") if stream_req is None: stream_req = True stream_req = bool(stream_req) tools_req = payload.get("tools") if isinstance(payload.get("tools"), list) else [] tools_responses = convert_tools_chat_to_responses(normalize_ollama_tools(tools_req)) - tool_choice = payload.get("tool_choice", "auto") + tool_choice = convert_tool_choice_chat_to_responses(payload.get("tool_choice", "auto")) parallel_tool_calls = bool(payload.get("parallel_tool_calls", False)) + text_format = convert_ollama_format_to_text_format(payload.get("format")) + + if ( + ollama_format_requests_json(payload.get("format")) + and (reasoning_compat or "").strip().lower() == "think-tags" + ): + # think tags would be prepended to content the caller asked to be json + reasoning_compat = "legacy" + # Passthrough Responses API tools (web_search) via ChatMock extension fields extra_tools: List[Dict[str, Any]] = [] had_responses_tools = False @@ -271,6 +283,7 @@ def ollama_chat() -> Response: allowed_efforts=allowed_efforts_for_model(model), ), service_tier=service_tier_resolution.service_tier, + text_format=text_format, ) if error_resp is not None: if verbose: @@ -297,7 +310,7 @@ def ollama_chat() -> Response: if verbose: print("[Passthrough] Upstream rejected tools; retrying without extras (args redacted)") base_tools_only = convert_tools_chat_to_responses(normalize_ollama_tools(tools_req)) - safe_choice = payload.get("tool_choice", "auto") + safe_choice = convert_tool_choice_chat_to_responses(payload.get("tool_choice", "auto")) upstream2, err2 = start_upstream_request( normalize_model_name(model, current_app.config.get("DEBUG_MODEL")), input_items, @@ -311,6 +324,7 @@ def ollama_chat() -> Response: allowed_efforts=allowed_efforts_for_model(model), ), service_tier=service_tier_resolution.service_tier, + text_format=text_format, ) record_rate_limits_from_response(upstream2) if err2 is None and upstream2 is not None and upstream2.status_code < 400: @@ -333,7 +347,7 @@ def ollama_chat() -> Response: if stream_req: def _gen(): - compat = (current_app.config.get("REASONING_COMPAT", "think-tags") or "think-tags").strip().lower() + compat = (reasoning_compat or "think-tags").strip().lower() think_open = False think_closed = False saw_any_summary = False @@ -551,7 +565,7 @@ def _gen(): finally: upstream.close() - if (current_app.config.get("REASONING_COMPAT", "think-tags") or "think-tags").strip().lower() == "think-tags": + if (reasoning_compat or "think-tags").strip().lower() == "think-tags": rtxt_parts = [] if isinstance(reasoning_summary_text, str) and reasoning_summary_text.strip(): rtxt_parts.append(reasoning_summary_text) diff --git a/chatmock/routes_openai.py b/chatmock/routes_openai.py index 673e22f..882d34c 100644 --- a/chatmock/routes_openai.py +++ b/chatmock/routes_openai.py @@ -32,6 +32,10 @@ from .upstream import normalize_model_name, start_upstream_raw_request, start_upstream_request from .utils import ( convert_chat_messages_to_responses_input, + convert_response_format_to_text_format, + convert_tool_choice_chat_to_responses, + convert_usage_responses_to_chat, + response_format_requests_json, convert_tools_chat_to_responses, sse_translate_chat, sse_translate_text, @@ -136,18 +140,12 @@ def chat_completions() -> Response: _log_json("OUT POST /v1/chat/completions", err) return jsonify(err), 400 - if isinstance(messages, list): - sys_idx = next((i for i, m in enumerate(messages) if isinstance(m, dict) and m.get("role") == "system"), None) - if isinstance(sys_idx, int): - sys_msg = messages.pop(sys_idx) - content = sys_msg.get("content") if isinstance(sys_msg, dict) else "" - messages.insert(0, {"role": "user", "content": content}) is_stream = bool(payload.get("stream")) stream_options = payload.get("stream_options") if isinstance(payload.get("stream_options"), dict) else {} include_usage = bool(stream_options.get("include_usage", False)) tools_responses = convert_tools_chat_to_responses(payload.get("tools")) - tool_choice = payload.get("tool_choice", "auto") + tool_choice = convert_tool_choice_chat_to_responses(payload.get("tool_choice", "auto")) parallel_tool_calls = bool(payload.get("parallel_tool_calls", False)) responses_tools_payload = payload.get("responses_tools") if isinstance(payload.get("responses_tools"), list) else [] extra_tools: List[Dict[str, Any]] = [] @@ -210,6 +208,16 @@ def chat_completions() -> Response: if tier_error is not None: return tier_error + text_format = convert_response_format_to_text_format(payload.get("response_format")) + text_verbosity = payload.get("verbosity") if isinstance(payload.get("verbosity"), str) else None + + if ( + response_format_requests_json(payload.get("response_format")) + and (reasoning_compat or "").strip().lower() == "think-tags" + ): + # think tags would be prepended to content the caller asked to be json + reasoning_compat = "legacy" + upstream, error_resp = start_upstream_request( model, input_items, @@ -218,6 +226,8 @@ def chat_completions() -> Response: parallel_tool_calls=parallel_tool_calls, reasoning_param=reasoning_param, service_tier=service_tier, + text_format=text_format, + text_verbosity=text_verbosity, ) if error_resp is not None: if verbose: @@ -246,7 +256,7 @@ def chat_completions() -> Response: if verbose: print("[Passthrough] Upstream rejected tools; retrying without extra tools (args redacted)") base_tools_only = convert_tools_chat_to_responses(payload.get("tools")) - safe_choice = payload.get("tool_choice", "auto") + safe_choice = convert_tool_choice_chat_to_responses(payload.get("tool_choice", "auto")) upstream2, err2 = start_upstream_request( model, input_items, @@ -255,6 +265,8 @@ def chat_completions() -> Response: parallel_tool_calls=parallel_tool_calls, reasoning_param=reasoning_param, service_tier=service_tier, + text_format=text_format, + text_verbosity=text_verbosity, ) record_rate_limits_from_response(upstream2) if err2 is None and upstream2 is not None and upstream2.status_code < 400: @@ -306,19 +318,10 @@ def chat_completions() -> Response: response_id = "chatcmpl" tool_calls: List[Dict[str, Any]] = [] error_message: str | None = None - usage_obj: Dict[str, int] | None = None + usage_obj: Dict[str, Any] | None = None - def _extract_usage(evt: Dict[str, Any]) -> Dict[str, int] | None: - try: - usage = (evt.get("response") or {}).get("usage") - if not isinstance(usage, dict): - return None - pt = int(usage.get("input_tokens") or 0) - ct = int(usage.get("output_tokens") or 0) - tt = int(usage.get("total_tokens") or (pt + ct)) - return {"prompt_tokens": pt, "completion_tokens": ct, "total_tokens": tt} - except Exception: - return None + def _extract_usage(evt: Dict[str, Any]) -> Dict[str, Any] | None: + return convert_usage_responses_to_chat((evt.get("response") or {}).get("usage")) try: for raw in upstream.iter_lines(decode_unicode=False): if not raw: @@ -503,18 +506,9 @@ def completions() -> Response: full_text = "" response_id = "cmpl" - usage_obj: Dict[str, int] | None = None - def _extract_usage(evt: Dict[str, Any]) -> Dict[str, int] | None: - try: - usage = (evt.get("response") or {}).get("usage") - if not isinstance(usage, dict): - return None - pt = int(usage.get("input_tokens") or 0) - ct = int(usage.get("output_tokens") or 0) - tt = int(usage.get("total_tokens") or (pt + ct)) - return {"prompt_tokens": pt, "completion_tokens": ct, "total_tokens": tt} - except Exception: - return None + usage_obj: Dict[str, Any] | None = None + def _extract_usage(evt: Dict[str, Any]) -> Dict[str, Any] | None: + return convert_usage_responses_to_chat((evt.get("response") or {}).get("usage")) try: for raw_line in upstream.iter_lines(decode_unicode=False): if not raw_line: diff --git a/chatmock/transform.py b/chatmock/transform.py index 7c611fb..525c1e6 100644 --- a/chatmock/transform.py +++ b/chatmock/transform.py @@ -147,3 +147,20 @@ def normalize_ollama_tools(tools: List[Dict[str, Any]] | None) -> List[Dict[str, ) return out + +def ollama_format_requests_json(fmt: Any) -> bool: + if isinstance(fmt, str): + return fmt.strip().lower() == "json" + return isinstance(fmt, dict) and bool(fmt) + + +def convert_ollama_format_to_text_format(fmt: Any) -> Dict[str, Any] | None: + """Map Ollama's format, when it holds a schema, onto a Responses text.format. + + The bare "json" string is left alone: upstream refuses json_object unless + the input mentions "json". + """ + + if not isinstance(fmt, dict) or not fmt: + return None + return {"type": "json_schema", "name": "response", "schema": fmt, "strict": False} diff --git a/chatmock/upstream.py b/chatmock/upstream.py index 9ad7941..dacc0f1 100644 --- a/chatmock/upstream.py +++ b/chatmock/upstream.py @@ -35,6 +35,8 @@ def start_upstream_request( parallel_tool_calls: bool = False, reasoning_param: Dict[str, Any] | None = None, service_tier: str | None = None, + text_format: Dict[str, Any] | None = None, + text_verbosity: str | None = None, ): access_token, account_id = get_effective_chatgpt_auth() if not access_token or not account_id: @@ -71,7 +73,7 @@ def start_upstream_request( "model": model, "input": input_items, "tools": tools or [], - "tool_choice": tool_choice if tool_choice in ("auto", "none") or isinstance(tool_choice, dict) else "auto", + "tool_choice": tool_choice if tool_choice in ("auto", "none", "required") or isinstance(tool_choice, dict) else "auto", "parallel_tool_calls": bool(parallel_tool_calls), "store": False, "stream": True, @@ -86,6 +88,13 @@ def start_upstream_request( responses_payload["reasoning"] = reasoning_param if isinstance(service_tier, str) and service_tier.strip(): responses_payload["service_tier"] = service_tier.strip().lower() + text: Dict[str, Any] = {} + if isinstance(text_format, dict) and text_format: + text["format"] = text_format + if isinstance(text_verbosity, str) and text_verbosity.strip(): + text["verbosity"] = text_verbosity.strip().lower() + if text: + responses_payload["text"] = text return start_upstream_raw_request( responses_payload, diff --git a/chatmock/utils.py b/chatmock/utils.py index 96dd314..c687a4a 100644 --- a/chatmock/utils.py +++ b/chatmock/utils.py @@ -145,8 +145,6 @@ def _normalize_image_data_url(url: str) -> str: input_items: List[Dict[str, Any]] = [] for message in messages: role = message.get("role") - if role == "system": - continue if role == "tool": call_id = message.get("tool_call_id") or message.get("id") @@ -213,11 +211,66 @@ def _normalize_image_data_url(url: str) -> str: if not content_items: continue - role_out = "assistant" if role == "assistant" else "user" + # Upstream refuses `system` but honours `developer`, so instructions + # keep their authority rather than arriving as one more user turn. + if role == "assistant": + role_out = "assistant" + elif role in ("system", "developer"): + role_out = "developer" + else: + role_out = "user" input_items.append({"type": "message", "role": role_out, "content": content_items}) return input_items +def convert_tool_choice_chat_to_responses(tool_choice: Any) -> Any: + """Map a Chat Completions tool_choice onto the flat Responses spelling.""" + + if not isinstance(tool_choice, dict): + return tool_choice + kind = tool_choice.get("type") + if kind == "function": + fn = tool_choice.get("function") + if isinstance(fn, dict) and isinstance(fn.get("name"), str): + return {"type": "function", "name": fn["name"]} + return tool_choice + if kind == "allowed_tools" and isinstance(tool_choice.get("allowed_tools"), dict): + body = tool_choice["allowed_tools"] + out: Dict[str, Any] = {"type": "allowed_tools"} + if isinstance(body.get("mode"), str): + out["mode"] = body["mode"] + tools = body.get("tools") + if isinstance(tools, list): + out["tools"] = [convert_tool_choice_chat_to_responses(t) for t in tools if isinstance(t, dict)] + return out + return tool_choice + + +def convert_usage_responses_to_chat(usage: Any) -> Dict[str, Any] | None: + """Map a Responses usage object onto the Chat Completions usage shape.""" + + if not isinstance(usage, dict): + return None + try: + prompt_tokens = int(usage.get("input_tokens") or 0) + completion_tokens = int(usage.get("output_tokens") or 0) + total_tokens = int(usage.get("total_tokens") or (prompt_tokens + completion_tokens)) + except Exception: + return None + out: Dict[str, Any] = { + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "total_tokens": total_tokens, + } + input_details = usage.get("input_tokens_details") + if isinstance(input_details, dict) and isinstance(input_details.get("cached_tokens"), int): + out["prompt_tokens_details"] = {"cached_tokens": input_details["cached_tokens"]} + output_details = usage.get("output_tokens_details") + if isinstance(output_details, dict) and isinstance(output_details.get("reasoning_tokens"), int): + out["completion_tokens_details"] = {"reasoning_tokens": output_details["reasoning_tokens"]} + return out + + def convert_tools_chat_to_responses(tools: Any) -> List[Dict[str, Any]]: out: List[Dict[str, Any]] = [] if not isinstance(tools, list): @@ -247,6 +300,46 @@ def convert_tools_chat_to_responses(tools: Any) -> List[Dict[str, Any]]: return out +def convert_response_format_to_text_format(response_format: Any) -> Dict[str, Any] | None: + """Map a Chat Completions response_format onto a Responses text.format.""" + + if not isinstance(response_format, dict): + return None + + # json_object is deliberately not mapped: upstream rejects it unless the + # input mentions "json", so honouring it would 400 requests that pass today. + if response_format.get("type") != "json_schema": + return None + + spec = response_format.get("json_schema") + if not isinstance(spec, dict): + spec = response_format + schema = spec.get("schema") + if not isinstance(schema, dict) or not schema: + return None + + name = spec.get("name") + text_format: Dict[str, Any] = { + "type": "json_schema", + "name": name.strip() if isinstance(name, str) and name.strip() else "response", + "schema": schema, + "strict": bool(spec.get("strict")), + } + description = spec.get("description") + if isinstance(description, str) and description.strip(): + text_format["description"] = description + return text_format + + +def response_format_requests_json(response_format: Any) -> bool: + """Whether the caller asked for JSON content, mapped upstream or not.""" + + return isinstance(response_format, dict) and response_format.get("type") in ( + "json_schema", + "json_object", + ) + + def load_chatgpt_tokens( ensure_fresh: bool = True, *, @@ -472,17 +565,8 @@ def _serialize_tool_args(eff_args: Any) -> str: else: return "{}" - def _extract_usage(evt: Dict[str, Any]) -> Dict[str, int] | None: - try: - usage = (evt.get("response") or {}).get("usage") - if not isinstance(usage, dict): - return None - pt = int(usage.get("input_tokens") or 0) - ct = int(usage.get("output_tokens") or 0) - tt = int(usage.get("total_tokens") or (pt + ct)) - return {"prompt_tokens": pt, "completion_tokens": ct, "total_tokens": tt} - except Exception: - return None + def _extract_usage(evt: Dict[str, Any]) -> Dict[str, Any] | None: + return convert_usage_responses_to_chat((evt.get("response") or {}).get("usage")) try: try: line_iterator = upstream.iter_lines(decode_unicode=False) @@ -828,17 +912,8 @@ def sse_translate_text(upstream, model: str, created: int, verbose: bool = False response_id = "cmpl-stream" upstream_usage = None - def _extract_usage(evt: Dict[str, Any]) -> Dict[str, int] | None: - try: - usage = (evt.get("response") or {}).get("usage") - if not isinstance(usage, dict): - return None - pt = int(usage.get("input_tokens") or 0) - ct = int(usage.get("output_tokens") or 0) - tt = int(usage.get("total_tokens") or (pt + ct)) - return {"prompt_tokens": pt, "completion_tokens": ct, "total_tokens": tt} - except Exception: - return None + def _extract_usage(evt: Dict[str, Any]) -> Dict[str, Any] | None: + return convert_usage_responses_to_chat((evt.get("response") or {}).get("usage")) try: for raw_line in upstream.iter_lines(decode_unicode=False): if not raw_line: diff --git a/tests/test_routes.py b/tests/test_routes.py index a490670..f577347 100644 --- a/tests/test_routes.py +++ b/tests/test_routes.py @@ -660,5 +660,552 @@ def close(self) -> None: ) +class ResponseFormatTests(unittest.TestCase): + SCHEMA = { + "type": "object", + "additionalProperties": False, + "required": ["summary"], + "properties": {"summary": {"type": "string"}}, + } + + def setUp(self) -> None: + reset_session_state() + self.app = create_app(model_sync=False) + self.client = self.app.test_client() + + def _post(self, mock_start, **extra) -> dict: + mock_start.return_value = ( + FakeUpstream( + [ + {"type": "response.output_text.delta", "delta": "{}"}, + {"type": "response.completed", "response": {"id": "resp-fmt"}}, + ] + ), + None, + ) + response = self.client.post( + "/v1/chat/completions", + json={"model": "gpt-5.6-sol", "messages": [{"role": "user", "content": "hi"}], **extra}, + ) + self.assertEqual(response.status_code, 200) + return mock_start.call_args.kwargs + + + + + @patch("chatmock.routes_openai.start_upstream_request") + def test_formats_that_are_not_forwarded(self, mock_start) -> None: + self.assertIsNone(self._post(mock_start)["text_format"]) + self.assertIsNone(self._post(mock_start, response_format={"type": "text"})["text_format"]) + self.assertIsNone( + self._post(mock_start, response_format={"type": "json_object"})["text_format"] + ) + self.assertIsNone( + self._post( + mock_start, response_format={"type": "json_schema", "json_schema": {"schema": {}}} + )["text_format"] + ) + + + @patch("chatmock.routes_openai.start_upstream_request") + def test_json_content_is_not_corrupted_by_think_tags(self, mock_start) -> None: + events = [ + {"type": "response.reasoning_summary_text.delta", "delta": "**Weighing it up**"}, + {"type": "response.reasoning_text.delta", "delta": "the full trace"}, + {"type": "response.output_text.delta", "delta": '{"summary":"ok"}'}, + {"type": "response.completed", "response": {"id": "resp-fmt"}}, + ] + + def post(**extra): + mock_start.return_value = (FakeUpstream(list(events)), None) + return self.client.post( + "/v1/chat/completions", + json={ + "model": "gpt-5.6-sol", + "messages": [{"role": "user", "content": "hi"}], + **extra, + }, + ).get_json()["choices"][0]["message"] + + self.assertTrue(post()["content"].startswith("")) + + message = post( + response_format={ + "type": "json_schema", + "json_schema": {"name": "r", "schema": self.SCHEMA}, + } + ) + self.assertEqual(json.loads(message["content"]), {"summary": "ok"}) + self.assertEqual(message["reasoning_summary"], "**Weighing it up**") + self.assertEqual(message["reasoning"], "the full trace") + + @patch("chatmock.routes_openai.start_upstream_request") + def test_non_think_tags_compat_is_left_alone(self, mock_start) -> None: + client = create_app(model_sync=False, reasoning_compat="o3").test_client() + mock_start.return_value = ( + FakeUpstream( + [ + {"type": "response.reasoning_summary_text.delta", "delta": "**Weighing it up**"}, + {"type": "response.output_text.delta", "delta": '{"summary":"ok"}'}, + {"type": "response.completed", "response": {"id": "resp-fmt"}}, + ] + ), + None, + ) + message = client.post( + "/v1/chat/completions", + json={ + "model": "gpt-5.6-sol", + "messages": [{"role": "user", "content": "hi"}], + "response_format": { + "type": "json_schema", + "json_schema": {"name": "r", "schema": self.SCHEMA}, + }, + }, + ).get_json()["choices"][0]["message"] + self.assertEqual( + message["reasoning"], {"content": [{"type": "text", "text": "**Weighing it up**"}]} + ) + self.assertEqual(json.loads(message["content"]), {"summary": "ok"}) + + def _post_ollama(self, mock_start, **extra) -> dict: + mock_start.return_value = ( + FakeUpstream( + [ + {"type": "response.reasoning_summary_text.delta", "delta": "**Weighing it up**"}, + {"type": "response.output_text.delta", "delta": '{"summary":"ok"}'}, + {"type": "response.completed", "response": {"id": "resp-fmt"}}, + ] + ), + None, + ) + response = self.client.post( + "/api/chat", + json={ + "model": "gpt-5.6-sol", + "messages": [{"role": "user", "content": "hi"}], + "stream": False, + **extra, + }, + ) + self.assertEqual(response.status_code, 200) + return response.get_json() + + + @patch("chatmock.routes_ollama.start_upstream_request") + def test_ollama_bare_json_is_uncorrupted_but_not_forwarded(self, mock_start) -> None: + body = self._post_ollama(mock_start, format="json") + self.assertIsNone(mock_start.call_args.kwargs["text_format"]) + self.assertEqual(json.loads(body["message"]["content"]), {"summary": "ok"}) + + + +class StructuredOutputWireTests(unittest.TestCase): + """Asserts past the start_upstream_request boundary, onto the actual payload. + + The route tests mock that call, so they pass whether or not the schema is + ever put on the wire. + """ + + SCHEMA = { + "type": "object", + "additionalProperties": False, + "required": ["summary"], + "properties": {"summary": {"type": "string"}}, + } + + EVENTS = [ + {"type": "response.reasoning_summary_text.delta", "delta": "**T**"}, + {"type": "response.output_text.delta", "delta": '{"summary":"ok"}'}, + {"type": "response.completed", "response": {"id": "resp-wire"}}, + ] + + def setUp(self) -> None: + reset_session_state() + self.client = create_app(model_sync=False).test_client() + + def _send(self, mock_post, path: str, body: dict) -> tuple[dict, str]: + mock_post.return_value = FakeUpstream(list(self.EVENTS)) + response = self.client.post(path, json=body) + self.assertEqual(response.status_code, 200) + sent = mock_post.call_args.kwargs["json"] + return sent, response.get_data(as_text=True) + + @patch("chatmock.upstream.get_effective_chatgpt_auth", return_value=("token", "acct")) + @patch("chatmock.upstream.requests.post") + def test_schema_reaches_the_upstream_payload(self, mock_post, _auth) -> None: + sent, _ = self._send( + mock_post, + "/v1/chat/completions", + { + "model": "gpt-5.6-sol", + "messages": [{"role": "user", "content": "hi"}], + "response_format": { + "type": "json_schema", + "json_schema": {"name": "weather", "schema": self.SCHEMA}, + }, + }, + ) + self.assertEqual( + sent["text"], + { + "format": { + "type": "json_schema", + "name": "weather", + "schema": self.SCHEMA, + "strict": False, + } + }, + ) + + flat, _ = self._send( + mock_post, + "/v1/chat/completions", + { + "model": "gpt-5.6-sol", + "messages": [{"role": "user", "content": "hi"}], + "response_format": { + "type": "json_schema", + "name": "weather", + "schema": self.SCHEMA, + "strict": True, + }, + }, + ) + self.assertEqual(flat["text"]["format"]["strict"], True) + + @patch("chatmock.upstream.get_effective_chatgpt_auth", return_value=("token", "acct")) + @patch("chatmock.upstream.requests.post") + def test_ollama_schema_reaches_the_upstream_payload(self, mock_post, _auth) -> None: + sent, _ = self._send( + mock_post, + "/api/chat", + { + "model": "gpt-5.6-sol", + "stream": False, + "messages": [{"role": "user", "content": "hi"}], + "format": self.SCHEMA, + }, + ) + self.assertEqual(sent["text"]["format"]["schema"], self.SCHEMA) + + @patch("chatmock.upstream.get_effective_chatgpt_auth", return_value=("token", "acct")) + @patch("chatmock.upstream.requests.post") + def test_no_text_key_without_a_format(self, mock_post, _auth) -> None: + sent, _ = self._send( + mock_post, + "/v1/chat/completions", + {"model": "gpt-5.6-sol", "messages": [{"role": "user", "content": "hi"}]}, + ) + self.assertNotIn("text", sent) + + @patch("chatmock.upstream.get_effective_chatgpt_auth", return_value=("token", "acct")) + @patch("chatmock.upstream.requests.post") + def test_streaming_content_is_uncorrupted(self, mock_post, _auth) -> None: + for path, body, extract in ( + ( + "/v1/chat/completions", + { + "model": "gpt-5.6-sol", + "stream": True, + "messages": [{"role": "user", "content": "hi"}], + "response_format": { + "type": "json_schema", + "json_schema": {"name": "r", "schema": self.SCHEMA}, + }, + }, + lambda line: json.loads(line[len("data: ") :])["choices"][0]["delta"].get("content"), + ), + ( + "/api/chat", + { + "model": "gpt-5.6-sol", + "stream": True, + "messages": [{"role": "user", "content": "hi"}], + "format": self.SCHEMA, + }, + lambda line: json.loads(line)["message"].get("content"), + ), + ): + with self.subTest(path=path): + _, raw = self._send(mock_post, path, body) + self.assertNotIn("", raw) + chunks = [] + for line in raw.splitlines(): + if not line.strip() or line.startswith("data: [DONE]"): + continue + try: + chunks.append(extract(line) or "") + except (ValueError, KeyError, IndexError): + continue + self.assertEqual(json.loads("".join(chunks)), {"summary": "ok"}) + + @patch("chatmock.upstream.get_effective_chatgpt_auth", return_value=("token", "acct")) + @patch("chatmock.upstream.requests.post") + def test_schema_survives_the_rejected_tools_retry(self, mock_post, _auth) -> None: + mock_post.side_effect = [ + FakeUpstream(status_code=400, content=b'{"error":{"message":"nope"}}'), + FakeUpstream(list(self.EVENTS)), + ] + response = self.client.post( + "/v1/chat/completions", + json={ + "model": "gpt-5.6-sol", + "messages": [{"role": "user", "content": "hi"}], + "responses_tools": [{"type": "web_search"}], + "response_format": { + "type": "json_schema", + "json_schema": {"name": "r", "schema": self.SCHEMA}, + }, + }, + ) + self.assertEqual(response.status_code, 200) + self.assertEqual(mock_post.call_count, 2) + retried = mock_post.call_args_list[1].kwargs["json"] + self.assertEqual(retried["text"]["format"]["schema"], self.SCHEMA) + + @patch("chatmock.upstream.get_effective_chatgpt_auth", return_value=("token", "acct")) + @patch("chatmock.upstream.requests.post") + def test_streaming_keeps_think_tags_without_a_format(self, mock_post, _auth) -> None: + _, raw = self._send( + mock_post, + "/v1/chat/completions", + { + "model": "gpt-5.6-sol", + "stream": True, + "messages": [{"role": "user", "content": "hi"}], + }, + ) + self.assertIn("", raw) + + if __name__ == "__main__": unittest.main() + + +class ChatRouteFidelityTests(unittest.TestCase): + """The chat-compat routes carry what the client sent, in the spelling upstream reads.""" + + EVENTS = [ + {"type": "response.output_text.delta", "delta": "ok"}, + { + "type": "response.completed", + "response": { + "id": "resp-fidelity", + "usage": { + "input_tokens": 1200, + "input_tokens_details": {"cached_tokens": 1024, "cache_write_tokens": 0}, + "output_tokens": 30, + "output_tokens_details": {"reasoning_tokens": 25}, + "total_tokens": 1230, + }, + }, + }, + ] + + def setUp(self) -> None: + reset_session_state() + self.client = create_app(model_sync=False).test_client() + + def _send(self, mock_post, path: str, body: dict, *, events: list | None = None) -> tuple[dict, str]: + mock_post.return_value = FakeUpstream(list(events if events is not None else self.EVENTS)) + response = self.client.post(path, json=body) + self.assertEqual(response.status_code, 200, response.get_data(as_text=True)) + return mock_post.call_args.kwargs["json"], response.get_data(as_text=True) + + @staticmethod + def _sse_payloads(text: str) -> list[dict]: + out = [] + for line in text.splitlines(): + if line.startswith("data: ") and line[6:].strip() not in ("", "[DONE]"): + out.append(json.loads(line[6:])) + return out + + @patch("chatmock.upstream.get_effective_chatgpt_auth", return_value=("token", "acct")) + @patch("chatmock.upstream.requests.post") + def test_system_and_developer_messages_keep_their_role(self, mock_post, _auth) -> None: + sent, _ = self._send( + mock_post, + "/v1/chat/completions", + { + "model": "gpt-5.6-sol", + "messages": [ + {"role": "system", "content": "You are terse."}, + {"role": "user", "content": "hi"}, + {"role": "developer", "content": [{"type": "text", "text": "Answer in French."}]}, + ], + }, + ) + self.assertEqual( + [(item["role"], item["content"][0]["type"]) for item in sent["input"]], + [("developer", "input_text"), ("user", "input_text"), ("developer", "input_text")], + ) + self.assertEqual(sent["input"][0]["content"][0]["text"], "You are terse.") + # Position is preserved rather than the system message being hoisted. + self.assertEqual(sent["input"][2]["content"][0]["text"], "Answer in French.") + + ollama, _ = self._send( + mock_post, + "/api/chat", + { + "model": "gpt-5.6-sol", + "stream": False, + "messages": [ + {"role": "system", "content": "You are terse."}, + {"role": "user", "content": "hi"}, + ], + }, + ) + self.assertEqual([item["role"] for item in ollama["input"]], ["developer", "user"]) + + @patch("chatmock.upstream.get_effective_chatgpt_auth", return_value=("token", "acct")) + @patch("chatmock.upstream.requests.post") + def test_tool_choice_is_spelled_the_responses_way(self, mock_post, _auth) -> None: + tools = [ + { + "type": "function", + "function": {"name": "get_time", "parameters": {"type": "object", "properties": {}}}, + } + ] + cases = [ + ({"type": "function", "function": {"name": "get_time"}}, {"type": "function", "name": "get_time"}), + ("required", "required"), + ("bogus", "auto"), + ( + { + "type": "allowed_tools", + "allowed_tools": { + "mode": "auto", + "tools": [{"type": "function", "function": {"name": "get_time"}}], + }, + }, + {"type": "allowed_tools", "mode": "auto", "tools": [{"type": "function", "name": "get_time"}]}, + ), + ] + for given, expected in cases: + with self.subTest(tool_choice=given): + sent, _ = self._send( + mock_post, + "/v1/chat/completions", + { + "model": "gpt-5.6-sol", + "messages": [{"role": "user", "content": "hi"}], + "tools": tools, + "tool_choice": given, + }, + ) + self.assertEqual(sent["tool_choice"], expected) + + @patch("chatmock.upstream.get_effective_chatgpt_auth", return_value=("token", "acct")) + @patch("chatmock.upstream.requests.post") + def test_verbosity_reaches_text(self, mock_post, _auth) -> None: + sent, _ = self._send( + mock_post, + "/v1/chat/completions", + {"model": "gpt-5.6-sol", "messages": [{"role": "user", "content": "hi"}], "verbosity": "low"}, + ) + self.assertEqual(sent["text"], {"verbosity": "low"}) + + both, _ = self._send( + mock_post, + "/v1/chat/completions", + { + "model": "gpt-5.6-sol", + "messages": [{"role": "user", "content": "hi"}], + "verbosity": "high", + "response_format": { + "type": "json_schema", + "json_schema": {"name": "r", "schema": {"type": "object", "properties": {}}}, + }, + }, + ) + self.assertEqual(both["text"]["verbosity"], "high") + self.assertEqual(both["text"]["format"]["name"], "r") + + @patch("chatmock.upstream.get_effective_chatgpt_auth", return_value=("token", "acct")) + @patch("chatmock.upstream.requests.post") + def test_usage_carries_cached_and_reasoning_token_details(self, mock_post, _auth) -> None: + expected = { + "prompt_tokens": 1200, + "completion_tokens": 30, + "total_tokens": 1230, + "prompt_tokens_details": {"cached_tokens": 1024}, + "completion_tokens_details": {"reasoning_tokens": 25}, + } + _, body = self._send( + mock_post, + "/v1/chat/completions", + {"model": "gpt-5.6-sol", "messages": [{"role": "user", "content": "hi"}]}, + ) + self.assertEqual(json.loads(body)["usage"], expected) + + _, stream = self._send( + mock_post, + "/v1/chat/completions", + { + "model": "gpt-5.6-sol", + "messages": [{"role": "user", "content": "hi"}], + "stream": True, + "stream_options": {"include_usage": True}, + }, + ) + usage_chunks = [chunk["usage"] for chunk in self._sse_payloads(stream) if chunk.get("usage")] + self.assertEqual(usage_chunks, [expected]) + + # An upstream that sends no detail objects yields the bare triple, not empty details. + bare_events = [ + {"type": "response.output_text.delta", "delta": "ok"}, + { + "type": "response.completed", + "response": {"id": "r", "usage": {"input_tokens": 5, "output_tokens": 1, "total_tokens": 6}}, + }, + ] + _, body = self._send( + mock_post, + "/v1/chat/completions", + {"model": "gpt-5.6-sol", "messages": [{"role": "user", "content": "hi"}]}, + events=bare_events, + ) + self.assertEqual( + json.loads(body)["usage"], {"prompt_tokens": 5, "completion_tokens": 1, "total_tokens": 6} + ) + + @patch("chatmock.upstream.get_effective_chatgpt_auth", return_value=("token", "acct")) + @patch("chatmock.upstream.requests.post") + def test_explicit_effort_is_forwarded_for_upstream_to_judge(self, mock_post, _auth) -> None: + # The catalog lists no `none` for gpt-5.6-luna; upstream accepts and honours it. + sent, _ = self._send( + mock_post, + "/v1/chat/completions", + { + "model": "gpt-5.6-luna", + "messages": [{"role": "user", "content": "hi"}], + "reasoning": {"effort": "none"}, + }, + ) + self.assertEqual(sent["reasoning"]["effort"], "none") + + # An effort upstream would not recognise at all is not forwarded. + sent, _ = self._send( + mock_post, + "/v1/chat/completions", + { + "model": "gpt-5.6-luna", + "messages": [{"role": "user", "content": "hi"}], + "reasoning": {"effort": "turbo"}, + }, + ) + self.assertEqual(sent["reasoning"]["effort"], "medium") + + def test_server_default_effort_is_still_clamped_to_the_model(self) -> None: + # No override: a server default the model does not list falls back to medium as before. + from chatmock.reasoning import build_reasoning_param + + self.assertEqual( + build_reasoning_param("ultra", "none", None, allowed_efforts=frozenset(("low", "medium", "high"))), + {"effort": "medium"}, + ) + self.assertEqual( + build_reasoning_param("ultra", "none", {"effort": "none"}, allowed_efforts=frozenset(("low", "medium"))), + {"effort": "none"}, + )