From c277fe20050674f4ade4e004a8dce06fb0a13610 Mon Sep 17 00:00:00 2001 From: gprocunier <49077358+gprocunier@users.noreply.github.com> Date: Fri, 31 Jul 2026 21:58:29 -0400 Subject: [PATCH] fix(llm): handle ordinary-only local model responses --- src/agent_loop.py | 110 ++++++++++- src/llm_core.py | 181 +++++++++++++++++- tests/test_agent_loop_malformed_output.py | 122 ++++++++++++ .../test_llm_core_ordinary_only_nonstream.py | 163 ++++++++++++++++ 4 files changed, 562 insertions(+), 14 deletions(-) create mode 100644 tests/test_agent_loop_malformed_output.py create mode 100644 tests/test_llm_core_ordinary_only_nonstream.py diff --git a/src/agent_loop.py b/src/agent_loop.py index cca93fe56..bc82c1fed 100644 --- a/src/agent_loop.py +++ b/src/agent_loop.py @@ -7,7 +7,9 @@ The LLM decides when to use tools by writing fenced code blocks. """ import asyncio +import base64 import collections +import hashlib import json import re import time @@ -3076,6 +3078,39 @@ def _detect_runaway_call(call_freq, threshold=15): return sig.split(":", 1)[0] if sig else None +def _stream_error_code(chunk: str) -> str: + """Return a stable stream error code without exposing error text.""" + if not isinstance(chunk, str) or not chunk.startswith("event: error"): + return "" + for line in chunk.splitlines(): + if not line.startswith("data:"): + continue + try: + payload = json.loads(line[5:].strip()) + except (TypeError, ValueError, json.JSONDecodeError): + return "" + if isinstance(payload, dict): + code = payload.get("code") + return code if isinstance(code, str) else "" + return "" + + +def _strip_reasoning_history_for_malformed_retry(messages: List[Dict]) -> List[Dict]: + """Drop provider-specific reasoning auxiliaries before one clean retry.""" + reasoning_fields = ("reasoning_content", "reasoning", "thinking") + cleaned = [] + for message in messages: + if not isinstance(message, dict): + cleaned.append(message) + continue + copy = dict(message) + if copy.get("role") == "assistant": + for field in reasoning_fields: + copy.pop(field, None) + cleaned.append(copy) + return cleaned + + async def stream_agent_loop( endpoint_url: str, model: str, @@ -3840,6 +3875,9 @@ async def stream_agent_loop( # lets a legit batch (e.g. 18 calendar events at once) through. _call_freq: collections.Counter = collections.Counter() _force_answer = False # set by loop-breaker → next round runs with NO tools + _malformed_retry_used = False + _retry_next_round_deterministic = False + _continue_truncated_output = False # Supervisor: how many times we've nudged the model after it announced # an action without emitting the tool call. Capped to prevent a model # that *can't* call the tool from looping forever. @@ -3878,6 +3916,10 @@ async def stream_agent_loop( _exhausted_rounds = False for round_num in range(1, max_rounds + 1): + continuation_round = _continue_truncated_output + _continue_truncated_output = False + _round_output_truncated = False + _retry_malformed_stream = False round_response = "" round_reasoning = "" # reasoning_content deltas (DeepSeek-thinking, vLLM --reasoning-parser) native_tool_calls = [] # populated if model uses function calling @@ -3894,7 +3936,7 @@ async def stream_agent_loop( # Merge native tool schemas with MCP tool schemas, filtering out # Only send function schemas for API models (OpenAI, Anthropic, etc.). # Local models use fenced code blocks or — schemas add overhead. - if _force_answer: + if _force_answer or continuation_round: # Loop-breaker decided the model has enough info but keeps # calling tools. Send NO tools this round so it's forced to # write the answer instead of flailing further. @@ -3969,14 +4011,16 @@ async def stream_agent_loop( bool(all_tool_schemas), agent_stream_timeout, ) + round_temperature = 0.0 if _retry_next_round_deterministic else temperature + _retry_next_round_deterministic = False async for chunk in stream_llm_with_fallback( _candidates, messages, - temperature=temperature, + temperature=round_temperature, max_tokens=max_tokens, prompt_type=prompt_type if round_num == 1 else None, tools=all_tool_schemas if all_tool_schemas else None, - tool_choice_none=_ody_doc_finetune_mode, + tool_choice_none=(_ody_doc_finetune_mode or _force_answer or continuation_round), timeout=agent_stream_timeout, session_id=session_id, workload=workload, @@ -3999,11 +4043,31 @@ async def stream_agent_loop( break # Forward error events from stream_llm to the frontend if chunk.startswith("event: error"): + error_code = _stream_error_code(chunk) + if error_code == "malformed_output": + if not _malformed_retry_used and round_num < max_rounds: + _malformed_retry_used = True + _retry_next_round_deterministic = True + messages[:] = _strip_reasoning_history_for_malformed_retry(messages) + _retry_malformed_stream = True + logger.warning( + "[agent-timing] malformed_output_retry round=%s elapsed=%.3fs", + round_num, + time.time() - _round_start, + ) + break + logger.warning( + "[agent-timing] malformed_output_terminal round=%s elapsed=%.3fs", + round_num, + time.time() - _round_start, + ) + yield chunk + return logger.warning( - "[agent-timing] stream_error round=%s elapsed=%.3fs chunk=%r", + "[agent-timing] stream_error round=%s elapsed=%.3fs code=%s", round_num, time.time() - _round_start, - chunk[:500], + error_code or "unspecified", ) yield chunk continue @@ -4012,7 +4076,9 @@ async def stream_agent_loop( data = json.loads(chunk[6:]) # IMPORTANT: check type-based events BEFORE "delta" key, # because tool_call_delta also has an "arg_delta" field. - if data.get("type") == "tool_call_delta": + if data.get("type") == "output_truncated": + _round_output_truncated = data.get("reason") == "length" + elif data.get("type") == "tool_call_delta": if tool_policy and tool_policy.blocks(data.get("name")): continue # Stream document content to frontend as AI generates it @@ -4192,6 +4258,38 @@ async def stream_agent_loop( _round_first_event_logged, _round_first_token_logged, ) + if _retry_malformed_stream: + continue + + if ( + _round_output_truncated + and round_response + and not native_tool_calls + and round_num < max_rounds + ): + logger.info( + "[agent] continuing truncated output chars=%d sha256=%s", + len(round_response), + hashlib.sha256(round_response.encode("utf-8", errors="replace")).hexdigest(), + ) + encoded_partial = base64.b64encode( + round_response.encode("utf-8", errors="replace") + ).decode("ascii") + messages.append(untrusted_context_message( + "base64-encoded partial model output", + encoded_partial, + )) + messages.append({ + "role": "user", + "content": ( + "The preceding untrusted data is a base64-encoded UTF-8 partial " + "model answer. Continue immediately after that answer. Do not " + "repeat prior text and do not call tools." + ), + }) + _continue_truncated_output = True + continue + _normalized_doc_round = ( _normalize_stream_document_fences( round_response, diff --git a/src/llm_core.py b/src/llm_core.py index 3e84c1060..2a86addee 100644 --- a/src/llm_core.py +++ b/src/llm_core.py @@ -278,6 +278,54 @@ def _stream_delta_event(text: str, *, thinking: bool = False) -> str: _DEGENERATE_WORD_RE = re.compile(r"[A-Za-z0-9_\u0370-\u03ff\u0400-\u04ff]+") +def _ordinary_only_nonstream_enabled(model: str) -> bool: + """Return whether *model* is explicitly allowlisted for non-stream mode.""" + configured = os.getenv("ODYSSEUS_NONSTREAM_ORDINARY_ONLY_MODELS", "") + allowed = { + item.strip().casefold() + for item in configured.split(",") + if item.strip() + } + return bool(model and model.strip().casefold() in allowed) + + +def _ordinary_only_nonstream_token_budget(max_tokens: int) -> int: + """Apply a small, bounded output budget to compatibility-mode requests.""" + configured = os.getenv("ODYSSEUS_NONSTREAM_ORDINARY_ONLY_MAX_TOKENS", "64") + try: + budget = int(configured) + except (TypeError, ValueError): + budget = 64 + budget = max(1, min(budget, 4096)) + if max_tokens and max_tokens > 0: + return min(max_tokens, budget) + return budget + + +def _malformed_output_event(reason: str, **metadata: int) -> str: + """Build a machine-readable error without replaying model-controlled text.""" + message = "Model returned malformed repetitive output. Retrying may help." + payload = { + "status": 502, + "text": message, + "error": message, + "code": "malformed_output", + "reason": reason, + } + payload.update(metadata) + return f"event: error\ndata: {json.dumps(payload)}\n\n" + + +def _looks_like_reserved_delimiter_output(text: str) -> bool: + """Detect a serialized special-token delimiter without naming any token.""" + if not isinstance(text, str): + return False + value = text.strip() + opening = chr(0x3C) + chr(0x7C) + closing = chr(0x7C) + chr(0x3E) + return value.startswith(opening) and closing in value[len(opening):] + + class _DegenerateStreamGuard: """Detect local-model token collapse before it floods the UI. @@ -293,11 +341,38 @@ class _DegenerateStreamGuard: self.same_run = 0 self.recent_tokens: List[str] = [] self.total_chars = 0 + self.last_visible_codepoint: Optional[int] = None + self.visible_codepoint_run = 0 def check(self, text: str) -> Optional[str]: if not text: return None self.total_chars += len(text) + for char in text: + if char.isspace(): + self.last_visible_codepoint = None + self.visible_codepoint_run = 0 + continue + codepoint = ord(char) + if codepoint == self.last_visible_codepoint: + self.visible_codepoint_run += 1 + else: + self.last_visible_codepoint = codepoint + self.visible_codepoint_run = 1 + if self.visible_codepoint_run >= 16: + logger.warning( + "[degenerate-stream] aborting model_hash=%s reason=%s run_length=%d codepoint=%d", + hashlib.sha256(self.model.encode("utf-8", errors="replace")).hexdigest()[:12], + "single_codepoint_run", + self.visible_codepoint_run, + codepoint, + ) + return _malformed_output_event( + "single_codepoint_run", + run_length=self.visible_codepoint_run, + codepoint=codepoint, + char_count=self.total_chars, + ) tokens = [t.lower() for t in _DEGENERATE_WORD_RE.findall(text) if len(t) >= 2] if not tokens: return None @@ -312,13 +387,16 @@ class _DegenerateStreamGuard: self.recent_tokens = self.recent_tokens[-96:] reason = None + metadata: Dict[str, int] = {} if self.same_run >= 28 and self.total_chars >= 100: - reason = f"repeated '{self.last_token}' {self.same_run} times" + reason = "same_token_run" + metadata = {"run_length": self.same_run, "char_count": self.total_chars} elif len(self.recent_tokens) >= 72: top = max(set(self.recent_tokens), key=self.recent_tokens.count) count = self.recent_tokens.count(top) if count >= 60 and count / max(len(self.recent_tokens), 1) >= 0.78: - reason = f"repeated '{top}' {count}/{len(self.recent_tokens)} recent tokens" + reason = "dominant_recent_token" + metadata = {"token_count": count, "window_size": len(self.recent_tokens)} if not reason and len(self.recent_tokens) >= 80: # Phrase loops are common on some local quantized MLX/MoE models: # "Also be a software developer mode?" repeated forever will not @@ -330,17 +408,19 @@ class _DegenerateStreamGuard: top_gram = max(set(grams), key=grams.count) gram_count = grams.count(top_gram) if gram_count >= 10: - reason = f"repeated phrase '{' '.join(top_gram)}' {gram_count} times" + reason = "repeated_token_phrase" + metadata = {"phrase_count": gram_count, "window_size": len(self.recent_tokens)} if not reason: return None - logger.warning("[degenerate-stream] aborting model=%s reason=%s", self.model, reason) - message = ( - f"Stopped generation: {self.model} started repeating tokens " - f"({reason}). Try a different model or lower temperature." + logger.warning( + "[degenerate-stream] aborting model_hash=%s reason=%s metadata=%s", + hashlib.sha256(self.model.encode("utf-8", errors="replace")).hexdigest()[:12], + reason, + metadata, ) - return f'event: error\ndata: {json.dumps({"status": 502, "text": message, "error": message})}\n\n' + return _malformed_output_event(reason, **metadata) def _model_activity_key(url: str, model: str) -> str: @@ -2212,6 +2292,7 @@ async def _stream_llm_inner(url: str, model: str, messages: List[Dict], temperat payload = _build_chatgpt_responses_payload(model, messages_copy, temperature, max_tokens, stream=True) else: target_url = _normalize_openai_chat_url(url) + ordinary_only_nonstream = _ordinary_only_nonstream_enabled(model) payload = { "model": model, "messages": messages_copy, @@ -2225,6 +2306,11 @@ async def _stream_llm_inner(url: str, model: str, messages: List[Dict], temperat if max_tokens and max_tokens > 0: tok_key = "max_completion_tokens" if _uses_max_completion_tokens(model) else "max_tokens" payload[tok_key] = max_tokens + if ordinary_only_nonstream: + payload["stream"] = False + payload.pop("stream_options", None) + tok_key = "max_completion_tokens" if _uses_max_completion_tokens(model) else "max_tokens" + payload[tok_key] = _ordinary_only_nonstream_token_budget(max_tokens) if tools: payload["tools"] = tools elif tool_choice_none: @@ -2537,6 +2623,85 @@ async def _stream_llm_inner(url: str, model: str, messages: List[Dict], temperat try: client = _get_http_client() h = await apply_kimi_code_headers_async(client, h, target_url) + if ordinary_only_nonstream: + response = await client.post( + target_url, + json=payload, + headers=h, + timeout=stream_timeout, + ) + _clear_host_dead(target_url) + if response.status_code != 200: + message = f"Upstream request failed with status {response.status_code}." + yield f'event: error\ndata: {json.dumps({"status": response.status_code, "text": message, "error": message})}\n\n' + return + + try: + body = response.json() + except (TypeError, ValueError, json.JSONDecodeError): + message = "Upstream returned an invalid structured response." + yield f'event: error\ndata: {json.dumps({"status": 502, "text": message, "error": message, "code": "invalid_upstream_response"})}\n\n' + return + + choices = body.get("choices") if isinstance(body, dict) else None + choice = choices[0] if isinstance(choices, list) and choices and isinstance(choices[0], dict) else {} + message_obj = choice.get("message") if isinstance(choice.get("message"), dict) else {} + content = message_obj.get("content") + content = content if isinstance(content, str) else "" + tool_calls = message_obj.get("tool_calls") + tool_calls = tool_calls if isinstance(tool_calls, list) else [] + finish_reason = choice.get("finish_reason") + + if content: + malformed = ( + _malformed_output_event("reserved_delimiter") + if _looks_like_reserved_delimiter_output(content) + else degenerate_guard.check(content) + ) + if malformed: + yield malformed + return + content = _strip_visible_chat_template_artifacts(content) + if content: + yield _stream_delta_event(content) + + normalized_calls = [] + for index, tool_call in enumerate(tool_calls): + if not isinstance(tool_call, dict): + continue + function = tool_call.get("function") + function = function if isinstance(function, dict) else {} + normalized_calls.append({ + "id": tool_call.get("id") or f"call_{index}", + "name": function.get("name") or "", + "arguments": function.get("arguments") or "{}", + }) + if normalized_calls: + yield f'data: {json.dumps({"type": "tool_calls", "calls": normalized_calls})}\n\n' + + usage = body.get("usage") if isinstance(body, dict) else None + if isinstance(usage, dict): + usage_data = { + "input_tokens": usage.get("prompt_tokens", 0), + "output_tokens": usage.get("completion_tokens", 0), + } + yield f'data: {json.dumps({"type": "usage", "data": usage_data})}\n\n' + + if finish_reason == "length" and content: + logger.info( + "[ordinary-only-nonstream] output truncated chars=%d sha256=%s", + len(content), + hashlib.sha256(content.encode("utf-8", errors="replace")).hexdigest(), + ) + yield f'data: {json.dumps({"type": "output_truncated", "reason": "length"})}\n\n' + elif not content and not normalized_calls: + message = "Model returned no ordinary response content." + yield f'event: error\ndata: {json.dumps({"status": 502, "text": message, "error": message, "code": "missing_ordinary_content"})}\n\n' + return + + yield "data: [DONE]\n\n" + return + async with client.stream('POST', target_url, json=payload, headers=h, timeout=stream_timeout) as r: _clear_host_dead(target_url) if r.status_code != 200: diff --git a/tests/test_agent_loop_malformed_output.py b/tests/test_agent_loop_malformed_output.py new file mode 100644 index 000000000..4d28c89dd --- /dev/null +++ b/tests/test_agent_loop_malformed_output.py @@ -0,0 +1,122 @@ +import asyncio +import json + +import src.agent_loop as agent_loop + + +def _collect(generator): + async def run(): + return [chunk async for chunk in generator] + + return asyncio.run(run()) + + +def _patch_common(monkeypatch): + monkeypatch.setattr(agent_loop, "get_setting", lambda key, default=None: default, raising=False) + monkeypatch.setattr(agent_loop, "get_mcp_manager", lambda: None, raising=False) + monkeypatch.setattr(agent_loop, "estimate_tokens", lambda *_args, **_kwargs: 10, raising=False) + + +def _data_payloads(chunks): + payloads = [] + for chunk in chunks: + for line in chunk.splitlines(): + if line.startswith("data:") and line[5:].strip() != "[DONE]": + payloads.append(json.loads(line[5:].strip())) + return payloads + + +def test_error_code_parser_uses_only_structured_code(): + event = "event: error\ndata: " + json.dumps({ + "status": 502, + "code": "malformed_output", + "text": chr(0x2603) * 24, + }) + "\n\n" + assert agent_loop._stream_error_code(event) == "malformed_output" + + +def test_reasoning_auxiliaries_are_removed_without_mutating_input(): + marker = chr(0x2603) * 24 + original = [{ + "role": "assistant", + "content": "visible", + "reasoning_content": marker, + "reasoning": marker, + "thinking": marker, + }] + cleaned = agent_loop._strip_reasoning_history_for_malformed_retry(original) + assert cleaned[0] == {"role": "assistant", "content": "visible"} + assert len(original[0]) == 5 + + +def test_malformed_output_retries_once_at_zero_temperature(monkeypatch): + _patch_common(monkeypatch) + calls = [] + + async def fake_stream(_candidates, messages, **kwargs): + calls.append({"temperature": kwargs["temperature"], "messages": list(messages)}) + if len(calls) == 1: + payload = { + "status": 502, + "code": "malformed_output", + "text": "Model returned malformed repetitive output.", + } + yield "event: error\ndata: " + json.dumps(payload) + "\n\n" + return + yield 'data: {"delta": "Recovered answer."}\n\n' + yield "data: [DONE]\n\n" + + monkeypatch.setattr(agent_loop, "stream_llm_with_fallback", fake_stream) + messages = [{ + "role": "assistant", + "content": "prior answer", + "reasoning": chr(0x2603) * 24, + }, {"role": "user", "content": "continue"}] + chunks = _collect(agent_loop.stream_agent_loop( + "http://localhost:8000/v1/chat/completions", + "example/model", + messages, + temperature=0.7, + max_rounds=2, + )) + + assert [call["temperature"] for call in calls] == [0.7, 0.0] + assert "reasoning" not in calls[1]["messages"][0] + assert not any(chunk.startswith("event: error") for chunk in chunks) + assert any(payload.get("delta") == "Recovered answer." for payload in _data_payloads(chunks)) + + +def test_valid_truncation_continues_without_tools_or_duplicate_output(monkeypatch): + _patch_common(monkeypatch) + calls = [] + + async def fake_stream(_candidates, messages, **kwargs): + calls.append({"messages": list(messages), **kwargs}) + if len(calls) == 1: + yield 'data: {"delta": "First section."}\n\n' + yield 'data: {"type": "output_truncated", "reason": "length"}\n\n' + yield "data: [DONE]\n\n" + return + yield 'data: {"delta": " Second section."}\n\n' + yield "data: [DONE]\n\n" + + monkeypatch.setattr(agent_loop, "stream_llm_with_fallback", fake_stream) + chunks = _collect(agent_loop.stream_agent_loop( + "http://localhost:8000/v1/chat/completions", + "example/model", + [{"role": "user", "content": "write a detailed answer"}], + max_rounds=2, + relevant_tools={"bash"}, + )) + payloads = _data_payloads(chunks) + + assert [payload.get("delta") for payload in payloads if "delta" in payload] == [ + "First section.", + " Second section.", + ] + assert not any(payload.get("type") == "output_truncated" for payload in payloads) + assert calls[1]["tools"] is None + assert calls[1]["tool_choice_none"] is True + assert "First section." not in calls[1]["messages"][-2]["content"] + assert "base64-encoded" in calls[1]["messages"][-1]["content"] + assert calls[1]["messages"][-2]["role"] == "user" diff --git a/tests/test_llm_core_ordinary_only_nonstream.py b/tests/test_llm_core_ordinary_only_nonstream.py new file mode 100644 index 000000000..cbedff1f0 --- /dev/null +++ b/tests/test_llm_core_ordinary_only_nonstream.py @@ -0,0 +1,163 @@ +import asyncio +import json + +from src import llm_core + + +class _Response: + status_code = 200 + + def __init__(self, body): + self._body = body + + def json(self): + return self._body + + +class _Client: + def __init__(self, body): + self.body = body + self.payload = None + + async def post(self, url, **kwargs): + self.payload = kwargs["json"] + return _Response(self.body) + + def stream(self, *args, **kwargs): + raise AssertionError("allowlisted compatibility mode must not open an SSE stream") + + +def _event_payloads(chunks): + payloads = [] + for chunk in chunks: + for line in chunk.splitlines(): + if line.startswith("data:") and line[5:].strip() != "[DONE]": + payloads.append(json.loads(line[5:].strip())) + return payloads + + +def _drive(monkeypatch, body, *, model="moonshotai/Kimi-K3", max_tokens=4096): + client = _Client(body) + monkeypatch.setenv("ODYSSEUS_NONSTREAM_ORDINARY_ONLY_MODELS", "moonshotai/Kimi-K3") + monkeypatch.delenv("ODYSSEUS_NONSTREAM_ORDINARY_ONLY_MAX_TOKENS", raising=False) + monkeypatch.setattr(llm_core, "_get_http_client", lambda: client) + monkeypatch.setattr(llm_core, "_is_host_dead", lambda _url: False) + monkeypatch.setattr(llm_core, "note_model_activity", lambda *_args, **_kwargs: None) + monkeypatch.setattr(llm_core, "_clear_host_dead", lambda *_args, **_kwargs: None) + + async def run(): + chunks = [] + async for chunk in llm_core.stream_llm( + "http://localhost:8000/v1/chat/completions", + model, + [{"role": "user", "content": "hello"}], + max_tokens=max_tokens, + ): + chunks.append(chunk) + return chunks + + return client, asyncio.run(run()) + + +def test_allowlist_is_exact_case_insensitive(monkeypatch): + monkeypatch.setenv( + "ODYSSEUS_NONSTREAM_ORDINARY_ONLY_MODELS", + " example/other , MoonshotAI/Kimi-K3 ", + ) + assert llm_core._ordinary_only_nonstream_enabled("moonshotai/kimi-k3") + assert not llm_core._ordinary_only_nonstream_enabled("moonshotai/kimi-k3-extra") + + +def test_token_budget_is_bounded_and_respects_lower_request(monkeypatch): + monkeypatch.setenv("ODYSSEUS_NONSTREAM_ORDINARY_ONLY_MAX_TOKENS", "99999") + assert llm_core._ordinary_only_nonstream_token_budget(0) == 4096 + assert llm_core._ordinary_only_nonstream_token_budget(32) == 32 + monkeypatch.setenv("ODYSSEUS_NONSTREAM_ORDINARY_ONLY_MAX_TOKENS", "invalid") + assert llm_core._ordinary_only_nonstream_token_budget(4096) == 64 + + +def test_guard_detects_single_codepoint_runs_without_replaying_text(): + repeated = chr(0x2603) + guard = llm_core._DegenerateStreamGuard("example/model") + assert guard.check(repeated * 8) is None + event = guard.check(repeated * 8) + assert event is not None + assert repeated not in event + payload = _event_payloads([event])[0] + assert payload["code"] == "malformed_output" + assert payload["reason"] == "single_codepoint_run" + assert payload["codepoint"] == 0x2603 + assert payload["run_length"] == 16 + + +def test_guard_does_not_replay_repeated_words(): + guard = llm_core._DegenerateStreamGuard("example/model") + event = guard.check(("sample " * 30).strip()) + assert event is not None + assert "sample" not in event + assert _event_payloads([event])[0]["reason"] == "same_token_run" + + +def test_reserved_delimiter_shape_is_detected_without_literal_tokens(): + opening = chr(0x3C) + chr(0x7C) + closing = chr(0x7C) + chr(0x3E) + value = opening + "reserved" + closing + assert llm_core._looks_like_reserved_delimiter_output(value) + assert not llm_core._looks_like_reserved_delimiter_output("ordinary answer") + + +def test_nonstream_emits_only_ordinary_content_and_truncation(monkeypatch): + hidden_value = chr(0x2603) * 24 + body = { + "choices": [{ + "message": {"content": "A concise answer.", "reasoning": hidden_value}, + "finish_reason": "length", + }], + "usage": {"prompt_tokens": 11, "completion_tokens": 64}, + } + client, chunks = _drive(monkeypatch, body) + events = _event_payloads(chunks) + + assert client.payload["stream"] is False + assert "stream_options" not in client.payload + assert client.payload["max_tokens"] == 64 + assert {event.get("delta") for event in events if "delta" in event} == {"A concise answer."} + assert any(event.get("type") == "output_truncated" for event in events) + assert hidden_value not in "".join(chunks) + + +def test_malformed_content_precedes_truncation_and_is_not_replayed(monkeypatch): + repeated = chr(0x2603) * 16 + body = { + "choices": [{ + "message": {"content": repeated}, + "finish_reason": "length", + }], + } + _client, chunks = _drive(monkeypatch, body) + combined = "".join(chunks) + events = _event_payloads(chunks) + + assert repeated not in combined + assert not any("delta" in event for event in events) + assert not any(event.get("type") == "output_truncated" for event in events) + assert events[0]["code"] == "malformed_output" + + +def test_reserved_delimiter_content_is_not_replayed(monkeypatch): + opening = chr(0x3C) + chr(0x7C) + closing = chr(0x7C) + chr(0x3E) + value = opening + "reserved" + closing + body = {"choices": [{"message": {"content": value}, "finish_reason": "stop"}]} + _client, chunks = _drive(monkeypatch, body) + combined = "".join(chunks) + events = _event_payloads(chunks) + + assert value not in combined + assert events[0]["code"] == "malformed_output" + assert events[0]["reason"] == "reserved_delimiter" + + +def test_non_allowlisted_model_keeps_streaming(monkeypatch): + monkeypatch.setenv("ODYSSEUS_NONSTREAM_ORDINARY_ONLY_MODELS", "moonshotai/Kimi-K3") + assert not llm_core._ordinary_only_nonstream_enabled("example/other")