fix: escape lone surrogates when rendering the LLM context 🤖🤖🤖

src/nooa/_immutable_json.py

@@ -46,6 +46,27 @@ def freeze(value):
     raise TypeError(f"Native extensions must contain JSON values only, got {type(value).__name__}")
 
 
+def utf8_encodable(value) -> bool:
+    """Whether every string in frozen ``value``, keys included, encodes as UTF-8.
+
+    Read-only: native payloads are never rewritten, so a caller that finds a
+    lone surrogate must replay without them instead.
+    """
+    if isinstance(value, str):
+        if value.isascii():
+            return True
+        try:
+            value.encode("utf-8")
+        except UnicodeEncodeError:
+            return False
+        return True
+    if isinstance(value, Mapping):
+        return all(utf8_encodable(k) and utf8_encodable(v) for k, v in value.items())
+    if isinstance(value, tuple):
+        return all(utf8_encodable(item) for item in value)
+    return True
+
+
 def json_containers(value):
     """Allocate wire/persistence containers, borrowing immutable scalar leaves."""
     if isinstance(value, Mapping):

src/nooa/context_blocks/renderer.py

@@ -20,6 +20,8 @@
 from collections.abc import Callable
 from typing import TYPE_CHECKING, Any, NamedTuple
 
+from pydantic import BaseModel
+
 if TYPE_CHECKING:
     from nooa.config.truncation_config import FormatConfig
 
@@ -37,6 +39,7 @@
     RenderedMessage,
     ResolvedBlock,
     Role,
+    ToolCallInfo,
 )
 from nooa.context_blocks.utils import camel_to_snake
 
@@ -134,6 +137,63 @@ def _apply_context_total_limit(
     return rendered, len(to_evict)
 
 
+def utf8_safe_text(text: str) -> str:
+    """Return ``text`` with lone UTF-16 surrogates escaped as ``\\udXXX`` text."""
+    if text.isascii():
+        return text
+    try:
+        text.encode("utf-8")
+    except UnicodeEncodeError:
+        return text.encode("utf-8", "backslashreplace").decode("utf-8")
+    return text
+
+
+def _utf8_safe(value: Any) -> Any:
+    """Return ``value`` with lone UTF-16 surrogates escaped as ``\\udXXX`` text.
+
+    Python strings can hold lone surrogates (an agent writing ``"\\ud800"`` in
+    code, ``os.listdir`` on non-UTF-8 names), but UTF-8 cannot encode them, so a
+    single one anywhere in the context makes every request fail to serialize.
+    Walks strings, dicts, lists, tuples and pydantic models; anything that needs
+    no change is returned as the same object, so callers' data is never mutated.
+    """
+    if isinstance(value, str):
+        return utf8_safe_text(value)
+    if isinstance(value, dict):
+        items = [(_utf8_safe(k), _utf8_safe(v)) for k, v in value.items()]
+        if all(k is k0 and v is v0 for (k, v), (k0, v0) in zip(items, value.items(), strict=True)):
+            return value
+        return dict(items)
+    if isinstance(value, (list, tuple)):
+        safe = [_utf8_safe(v) for v in value]
+        if all(v is v0 for v, v0 in zip(safe, value, strict=True)):
+            return value
+        return safe if isinstance(value, list) else tuple(safe)
+    if isinstance(value, BaseModel):
+        update = {}
+        for name in type(value).model_fields:
+            # Native replay turns are resent verbatim; a turn whose content
+            # changed here is rendered from that content instead.
+            if name == "replay_message":
+                continue
+            field = getattr(value, name)
+            # Dict arguments reach the request and the journal only through
+            # json.dumps, whose ASCII escapes keep them exact and encodable.
+            if isinstance(value, ToolCallInfo) and name == "arguments" and isinstance(field, dict):
+                continue
+            safe = _utf8_safe(field)
+            if safe is not field:
+                update[name] = safe
+        return value.model_copy(update=update) if update else value
+    return value
+
+
+def _with_utf8_safe_content(block: ResolvedBlock) -> ResolvedBlock:
+    """Escape lone surrogates in ``block.content`` so budgets measure the sent text."""
+    content = _utf8_safe(block.content)
+    return block if content is block.content else block.model_copy(update={"content": content})
+
+
 def render_context(
     blocks: list[ResolvedBlock],
     *,
@@ -167,7 +227,7 @@ def render_context(
     count_fn: Callable[[str], int] = count_tokens if count_tokens is not None else len
 
     # Partition for truncation only.
-    system_blocks = [b for b in blocks if b.role == Role.SYSTEM]
+    system_blocks = [_with_utf8_safe_content(b) for b in blocks if b.role == Role.SYSTEM]
     message_blocks = [b for b in blocks if b.role != Role.SYSTEM]
 
     # Pre-serialize non-tool events so their content can be measured.
@@ -182,7 +242,7 @@ def render_context(
             )
             content = block_formatter.format_event(block.event, event_format=resolved_event_format)
             block = block.model_copy(update={"content": content})
-        serialized_messages.append(block)
+        serialized_messages.append(_with_utf8_safe_content(block))
     message_blocks = serialized_messages
 
     # Total-context eviction: mark over-budget blocks EVICTED in place.
@@ -214,8 +274,11 @@ def render_context(
         reserved_output_tokens=reserved_output_tokens,
     )
 
-    # Neutral message list → provider wire format.
-    messages = block_formatter.format([*system_blocks, *message_blocks])
+    # Neutral message list → provider wire format. Block content is already
+    # escaped (so the budget and stats above measure the sent text); this pass
+    # covers the remaining fields (tool calls, images, parts). Both the request
+    # and the tracing journal encode these messages.
+    messages = _utf8_safe(block_formatter.format([*system_blocks, *message_blocks]))
     output = provider_formatter.format(messages)
     return RenderResult(
         output=output,

src/nooa/unifiedllm/chat_parts.py

@@ -9,7 +9,7 @@
 
 from typing import Any
 
-from nooa._immutable_json import json_containers
+from nooa._immutable_json import json_containers, utf8_encodable
 from nooa.llm_types import AssistantPart, AssistantReasoning, AssistantText, LLMResponse, ToolCall
 from nooa.unifiedllm.cache_policy import wrap_anthropic_text
 
@@ -203,6 +203,12 @@ def project_chat_turn(
     know how those ids encode private state. No stored part is mutated.
     """
     compatible = scope is not None and turn.replay_scope == scope
+    if compatible and not all(utf8_encodable(part.native) for part in turn.parts):
+        logger.warning(
+            "Stored native state holds a lone surrogate that UTF-8 cannot encode; "
+            "replaying portable text instead of native state."
+        )
+        compatible = False
     provider = _scope_provider(scope)
     if turn.replay_scope and not compatible:
         logger.warning(

src/nooa/unifiedllm/response_parts.py

@@ -9,7 +9,7 @@
 import logging
 from typing import Any
 
-from nooa._immutable_json import freeze, json_containers
+from nooa._immutable_json import freeze, json_containers, utf8_encodable
 from nooa.llm_types import AssistantPart, AssistantReasoning, AssistantText, LLMResponse, ToolCall
 
 from .replay_state import (
@@ -235,6 +235,12 @@ def project_turn(
                 )
                 compatible = False
                 break
+    if compatible and not all(utf8_encodable(part.native) for part in turn.parts):
+        logger.warning(
+            "Stored native state holds a lone surrogate that UTF-8 cannot encode; "
+            "replaying portable text instead of native state."
+        )
+        compatible = False
     if turn.replay_scope and not compatible:
         logger.warning(
             "Incompatible assistant turn: replaying portable parts without native state."

tests/context_blocks/test_cached_renderer.py

@@ -321,3 +321,34 @@ def test_empty_input(self):
             provider_formatter=OpenAIProviderFormatter(),
         ).output
         assert result == []
+
+
+class TestCachedRendererLoneSurrogates:
+    """Escaping a lone surrogate keeps the cache layout and prefix stable."""
+
+    def _render(self, status: str) -> list:
+        return render_context(
+            [
+                _static_block("sys", "Stable prefix 🎉"),
+                _dynamic_block("todo_status", status, expr="self.todo.status()"),
+            ],
+            block_formatter=CachedBlockFormatter(),
+            provider_formatter=OpenAIProviderFormatter(),
+        ).output
+
+    def test_surrogate_in_dynamic_suffix_leaves_prefix_and_boundary_intact(self):
+        clean = self._render("note ok")
+        poisoned = self._render("note ::1%\ud800.x")
+
+        assert poisoned[0] == clean[0]
+        assert isinstance(poisoned[1], CacheBoundary)
+        assert "note ::1%\\ud800.x" in poisoned[2]["content"]
+        json.dumps(poisoned[2], ensure_ascii=False).encode("utf-8")
+
+    def test_escaping_is_deterministic_across_turns(self):
+        """The same poisoned context renders identically every turn, so it can be cached."""
+        first = self._render("note ::1%\ud800.x")
+        second = self._render("note ::1%\ud800.x")
+
+        assert first[0] == second[0]
+        assert first[2] == second[2]

tests/context_blocks/test_renderer.py

@@ -7,15 +7,17 @@
 evaluation, no class — just a pure function.
 """
 
+import json
+
 import pytest
 
 from nooa.context_blocks.formatter import (
     MarkdownBlockFormatter,
     OpenAIProviderFormatter,
     XMLBlockFormatter,
 )
-from nooa.context_blocks.models import BlockMetadata, ResolvedBlock, Role
-from nooa.context_blocks.renderer import render_context
+from nooa.context_blocks.models import BlockMetadata, RenderedMessage, ResolvedBlock, Role
+from nooa.context_blocks.renderer import _utf8_safe, render_context
 
 
 class TestRenderContextBasic:
@@ -638,3 +640,136 @@ def counter(s: str) -> int:
             count_tokens=counter,
         )
         assert len(call_count) > 0
+
+
+class TestRenderContextLoneSurrogates:
+    """A lone surrogate in the context must not make the request unencodable."""
+
+    def test_system_block_surrogate_is_escaped_in_output_and_messages(self):
+        """A status block holding "\\ud800" renders as the six characters ``\\ud800``."""
+        content = "Notes:\n10. probe ::1%\ud800.synthetic-secret passes"
+        block = ResolvedBlock(key="todo_status", content=content)
+
+        result = render_context(
+            [block],
+            block_formatter=XMLBlockFormatter(),
+            provider_formatter=OpenAIProviderFormatter(),
+        )
+
+        system_content = result.output[0]["content"]
+        system_content.encode("utf-8")
+        assert "::1%\\ud800.synthetic-secret" in system_content
+        assert result.messages[0].content == system_content
+        assert block.content == content
+
+    def test_tool_result_surrogate_is_escaped(self):
+        """Event-derived messages are made safe, not only context blocks."""
+        from nooa.context_blocks.events import ToolCallEvent, ToolResult
+
+        event = ToolCallEvent(
+            tool_call_id="call_1",
+            name="python_cell",
+            arguments={"code": "print(name)"},
+            result=ToolResult(tool_call_id="call_1", content="bad-\udcff-name"),
+        )
+        blocks = [ResolvedBlock(key="tool1", content="", role=Role.ASSISTANT, event=event)]
+
+        output = render_context(
+            blocks,
+            block_formatter=XMLBlockFormatter(),
+            provider_formatter=OpenAIProviderFormatter(),
+        ).output
+
+        json.dumps(output, ensure_ascii=False).encode("utf-8")
+        assert output[2]["content"] == "bad-\\udcff-name"
+        assert event.result.content == "bad-\udcff-name"
+
+    def test_replay_message_renders_escaped_content(self):
+        """A replay turn is handed the escaped content, and its object is not rewritten."""
+        seen = []
+
+        class Replay:
+            def render_message(self, content, tool_calls, reasoning=None):
+                seen.append(content)
+                return {"role": "assistant", "content": content}
+
+        replay = Replay()
+
+        class ReplayFormatter(XMLBlockFormatter):
+            def format(self, blocks):
+                return [
+                    RenderedMessage(role=Role.ASSISTANT, content="x\udc80y", replay_message=replay)
+                ]
+
+        result = render_context(
+            [],
+            block_formatter=ReplayFormatter(),
+            provider_formatter=OpenAIProviderFormatter(),
+        )
+
+        assert seen == ["x\\udc80y"]
+        assert result.output[0]["content"] == "x\\udc80y"
+        assert result.messages[0].replay_message is replay
+
+    def test_context_budget_measures_escaped_content(self):
+        """Eviction and stats see the escaped text, which is longer than the original."""
+        status = "\ud800" * 100  # 100 chars before escaping, 600 after
+        blocks = [
+            ResolvedBlock(key="persona", content="Be helpful."),
+            ResolvedBlock(
+                key="todo_status",
+                content=status,
+                metadata=BlockMetadata(user_block=True),
+            ),
+        ]
+
+        result = render_context(
+            blocks,
+            block_formatter=XMLBlockFormatter(),
+            provider_formatter=OpenAIProviderFormatter(),
+            context_limit=200,
+            count_tokens=len,
+        )
+
+        assert result.stats.context_blocks_dropped == 1
+        assert "EVICTED" in result.output[0]["content"]
+        assert "Be helpful." in result.output[0]["content"]
+
+    def test_dict_tool_arguments_are_left_to_json_encoding(self):
+        """Dict arguments keep their exact keys; json.dumps escapes them for the wire."""
+        from nooa.context_blocks.models import ToolCallInfo
+
+        arguments = {"\ud800": "lone", "\\ud800": "literal"}
+        message = RenderedMessage(
+            role=Role.ASSISTANT,
+            tool_calls=(ToolCallInfo(id="call_1", name="python_cell", arguments=arguments),),
+        )
+
+        class Formatter(XMLBlockFormatter):
+            def format(self, blocks):
+                return [message]
+
+        result = render_context(
+            [],
+            block_formatter=Formatter(),
+            provider_formatter=OpenAIProviderFormatter(),
+        )
+
+        assert result.messages[0].tool_calls[0].arguments == arguments
+        sent = result.output[0]["tool_calls"][0]["function"]["arguments"]
+        sent.encode("utf-8")
+        assert json.loads(sent) == arguments
+
+    def test_valid_unicode_is_returned_unchanged(self):
+        """Encodable text, emoji included, passes through as the same objects."""
+        blocks = [ResolvedBlock(key="persona", content="Héllo 🎉 — ok")]
+
+        result = render_context(
+            blocks,
+            block_formatter=XMLBlockFormatter(),
+            provider_formatter=OpenAIProviderFormatter(),
+        )
+
+        assert "Héllo 🎉 — ok" in result.output[0]["content"]
+        assert _utf8_safe(result.output) is result.output
+        assert _utf8_safe(result.messages) is result.messages

tests/test_immutable_json.py

@@ -0,0 +1,18 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Tests for frozen native JSON: lone surrogates are detected, never rewritten."""
+
+from nooa._immutable_json import freeze, json_containers, utf8_encodable
+
+
+def test_freeze_keeps_lone_surrogates_exact():
+    value = {"k\ud800": ["x\udc80", {"text": "ok \ud83d"}], "n": 1}
+
+    assert json_containers(freeze(value)) == value
+
+
+def test_utf8_encodable_finds_lone_surrogates_in_keys_and_values():
+    assert utf8_encodable(freeze({"a": ["Héllo 🎉", {"b": None}], "n": 1.5}))
+    assert utf8_encodable(None)
+    assert not utf8_encodable(freeze({"a": ("ok", {"b": "x\ud800"})}))
+    assert not utf8_encodable(freeze({"k\udcff": "v"}))