fix(llm): handle ordinary-only local model responses

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 <tool_code> — 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,

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:

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"

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")