feat(llm): retain compatibility-scoped OpenAI reasoning state

src/nooa/_llm_state.py

@@ -10,7 +10,6 @@
 
 from __future__ import annotations
 
-import copy
 from typing import Any
 from uuid import uuid4
 
@@ -32,7 +31,9 @@ def __init__(
         replay_batch_size: int = 0,
     ):
         super().__init__(message)
-        self.llm_state = copy.deepcopy(llm_state)
+        # Borrow immutable event state through to the provider adapter. The
+        # adapter owns serialization and must not mutate caller input.
+        self.llm_state = llm_state
         self.reasoning = reasoning
         self.replay_batch_id = replay_batch_id
         self.replay_batch_size = replay_batch_size
@@ -94,36 +95,17 @@ def demote_reasoning_text(message: dict[str, Any], reasoning: str | None) -> Non
         message["content"] = reasoning
 
 
-def demote_chat_reasoning(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    """Build public Chat messages, withholding opaque state by default."""
-    prepared: list[dict[str, Any]] = []
-    for original in messages:
-        state = carried_state(original)
-        reasoning = carried_reasoning(original)
-        message = copy.deepcopy(dict(original))
-        message.pop(LLM_STATE_KEY, None)
-        demote_reasoning_text(message, reasoning)
-        if (
-            state is not None
-            and not reasoning
-            and message.get("role") == "assistant"
-            and not message.get("content")
-            and not message.get("tool_calls")
-        ):
-            continue
-        prepared.append(message)
-    return prepared
-
-
 def demote_responses_batch(
     batch: list[dict[str, Any]],
     llm_state: dict[str, Any] | None,
     reasoning: str | None,
 ) -> list[dict[str, Any]]:
-    """Build public Responses items, withholding opaque state by default."""
-    clean = [copy.deepcopy(dict(item)) for item in batch]
-    for item in clean:
-        item.pop(LLM_STATE_KEY, None)
+    """Demote reasoning into a cleaned, request-owned Responses batch.
+
+    Replay preparation has already detached public data and removed untrusted
+    state. Reuse that batch instead of allocating another copy of the history.
+    """
+    clean = batch
 
     if not reasoning:
         if (

src/nooa/context_blocks/models.py

@@ -13,7 +13,7 @@
 
 from typing import Annotated, Any, Literal
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, SkipValidation
 
 # Import EventBase here (not forward ref) — possible because events.py imports
 # Role from roles.py, breaking the circular dependency.
@@ -300,10 +300,12 @@ class RenderedMessage(BaseModel):
         default_factory=tuple,
         description="Complete ordered tool-call batch on an assistant turn",
     )
-    llm_state: dict[str, Any] | None = Field(
+    llm_state: SkipValidation[dict[str, Any] | None] = Field(
         default=None,
         repr=False,
-        description="Opaque state carried only to the UnifiedLLM replay boundary",
+        description=(
+            "Borrowed immutable opaque state carried through the UnifiedLLM replay boundary"
+        ),
     )
     reasoning: str | None = Field(
         default=None,

src/nooa/nemo_relay_middleware.py

@@ -224,7 +224,12 @@ async def _wrapper(req: Any) -> Any:
         if hasattr(req, "content") and isinstance(req.content, dict):
             intercepted = req.content
             intercepted_msgs = intercepted.get("messages")
-            if intercepted_msgs is not None:
+            # Relay crosses a JSON boundary, so even a no-op request returns
+            # fresh plain dicts. Keep NOOA's original dict subclasses when the
+            # public payload is unchanged: their non-JSON sidecars carry opaque
+            # replay state to UnifiedLLM. A real intercept change replaces the
+            # originals and therefore drops private state fail-closed.
+            if intercepted_msgs is not None and intercepted_msgs != ctx.messages:
                 ctx.messages = intercepted_msgs
             # Propagate any supported param changes from the intercept.
             for key in _PROPAGATABLE_LLM_PARAMS:

src/nooa/runtime/middleware.py

@@ -26,7 +26,7 @@
 from collections.abc import Awaitable, Callable
 from typing import Any
 
-from pydantic import BaseModel, ConfigDict
+from pydantic import BaseModel, ConfigDict, SkipValidation
 
 from nooa.agent import Agent
 from nooa.events import ExecutionResult
@@ -104,7 +104,10 @@ class LLMCallContext(BaseModel):
 
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
-    messages: list[dict[str, Any]]
+    # Preserve in-memory message subclasses used for non-serializable provider
+    # replay state. Messages are runtime-produced and middleware intentionally
+    # receives a mutable list, so coercing each dict provides no safety here.
+    messages: SkipValidation[list[dict[str, Any]]]
     params: dict[str, Any] = {}
     agent: Agent | None = None
     runtime: ActorRuntime | None = None

src/nooa/tracing/_litellm_journal.py

@@ -36,6 +36,7 @@
 from opentelemetry import trace as otel_trace
 
 from nooa.tracing._journal_builder import _encode_image
+from nooa.tracing._secret_scrubber import scrub_value
 from nooa.tracing._session import get_session
 
 log = logging.getLogger(__name__)
@@ -69,6 +70,12 @@ def _msg_to_dict(msg: Any) -> dict:
         return {"raw": str(msg)}
 
 
+def _safe_msg_to_dict(msg: Any) -> dict:
+    """Normalize one provider message and remove issuer-only opaque state."""
+    scrubbed, _ = scrub_value(_msg_to_dict(msg))
+    return scrubbed if isinstance(scrubbed, dict) else {}
+
+
 def _extract_output_msgs(response_obj: Any) -> list[dict]:
     """Pull assistant messages out of a litellm completion response.
 
@@ -546,7 +553,7 @@ def log_pre_api_call(self, model: str, messages: list, kwargs: dict) -> None:
             # No sideband — publish the raw messages as the skeleton
             # with no block refs. The viewer just uses their content
             # as-is, matching what the wire shows.
-            input_skeleton = [_msg_to_dict(m) for m in messages]
+            input_skeleton = [_safe_msg_to_dict(m) for m in messages]
 
         span_id = self._current_span_id()
         with self._lock:
@@ -578,7 +585,9 @@ def log_success_event(
         # small and re-uses any hash that overlaps with messages the
         # agent will echo back on subsequent turns.
         output_blocks: dict[str, str] = {}
-        output_messages = [_skeleton_dict_message(m, output_blocks) for m in raw_output]
+        output_messages = [
+            _skeleton_dict_message(_safe_msg_to_dict(m), output_blocks) for m in raw_output
+        ]
         if output_blocks:
             self._send_new_blocks(session_id, output_blocks)
 

src/nooa/tracing/_secret_scrubber.py

@@ -15,6 +15,7 @@
     provider.add_span_processor(SecretScrubSpanProcessor(inner_processor))
 """
 
+import json
 import logging
 import re
 import threading
@@ -47,6 +48,11 @@
     }
 )
 
+# Provider-owned replay state is not a user credential, but it has the same
+# telemetry rule: it may go back to its issuer and nowhere else. Provider
+# adapters add their exact wire keys here as support is introduced.
+_OPAQUE_PROVIDER_STATE_KEYS = frozenset({"encrypted_content", "nooa_llm_state"})
+
 
 def _is_sensitive_key(key: Any) -> bool:
     normalized = re.sub(r"[^a-z0-9]+", "_", str(key).lower()).strip("_")
@@ -55,6 +61,19 @@ def _is_sensitive_key(key: Any) -> bool:
     )
 
 
+def _is_opaque_provider_state_key(key: Any) -> bool:
+    normalized = re.sub(r"[^a-z0-9]+", "_", str(key).lower()).strip("_")
+    return normalized in _OPAQUE_PROVIDER_STATE_KEYS
+
+
+def _redact_key(key: Any) -> str | None:
+    if _is_sensitive_key(key):
+        return "sensitive_key"
+    if _is_opaque_provider_state_key(key):
+        return "opaque_provider_state"
+    return None
+
+
 # ---------------------------------------------------------------------------
 # Regex-based secret patterns — high precision, low false positives
 # ---------------------------------------------------------------------------
@@ -214,19 +233,31 @@ def scrub_value(value: Any) -> tuple[Any, int]:
     number of secrets redacted within this value.
     """
     if isinstance(value, str):
-        return scrub_string(value)
+        # OpenInference records LLM inputs as JSON string span attributes.
+        # Only parse objects/arrays. Matching the prefix avoids allocating a
+        # stripped copy of a potentially large tool output just to inspect it.
+        if not re.match(r"\s*[\[{]", value):
+            return scrub_string(value)
+        try:
+            decoded = json.loads(value)
+        except (json.JSONDecodeError, TypeError):
+            return scrub_string(value)
+        decoded, json_count = scrub_value(decoded)
+        if json_count:
+            return json.dumps(decoded, separators=(",", ":"), ensure_ascii=False), json_count
+        return value, 0
     if isinstance(value, dict):
-        scrubbed: dict[Any, Any] = {}
+        scrubbed_mapping: dict[Any, Any] = {}
         count = 0
         for key, item in value.items():
-            if _is_sensitive_key(key):
-                scrubbed[key] = REDACTED
-                stats.record("sensitive_key")
+            if reason := _redact_key(key):
+                scrubbed_mapping[key] = REDACTED
+                stats.record(reason)
                 count += 1
             else:
-                scrubbed[key], n = scrub_value(item)
+                scrubbed_mapping[key], n = scrub_value(item)
                 count += n
-        return scrubbed, count
+        return scrubbed_mapping, count
     if isinstance(value, (list, tuple)):
         new_items = []
         count = 0
@@ -268,9 +299,9 @@ def on_end(self, span: ReadableSpan) -> None:
                 redacted_count = 0
 
                 for key, value in span.attributes.items():
-                    if _is_sensitive_key(key):
+                    if reason := _redact_key(key):
                         new_value, n = REDACTED, 1
-                        stats.record("sensitive_key")
+                        stats.record(reason)
                     else:
                         new_value, n = scrub_value(value)
                     scrubbed[key] = new_value

src/nooa/unifiedllm/fake.py

@@ -13,6 +13,8 @@
 
 from nooa.unifiedllm.unifiedllm import LLMResponse, LLMUsage, Tool, ToolCall, UnifiedLLM
 
+from .replay_state import prepare_chat_messages
+
 
 class FakeLLMClient(UnifiedLLM):
     """
@@ -81,7 +83,8 @@ async def acall(
         """
         async with self._lock:
             self.call_count += 1
-            self.last_messages = messages
+            # A non-provider test client must never observe private replay state.
+            self.last_messages = prepare_chat_messages(messages, None)
             self.last_tools = tools
 
             # Return next response from queue, or empty response if none left
@@ -108,7 +111,7 @@ def call(
         """Synchronous version of acall for UnifiedLLM compatibility."""
         # For sync call, we don't need locking since tests are usually single-threaded
         self.call_count += 1
-        self.last_messages = messages
+        self.last_messages = prepare_chat_messages(messages, None)
         self.last_tools = tools
 
         if self._response_queue:

src/nooa/unifiedllm/registry.py

@@ -38,6 +38,9 @@
         top_p: 1.0                           # optional
         max_tokens: 4096                     # optional
         drop_params: true                    # optional, defaults to true
+        store: false                         # optional Responses API control
+        include:                             # optional Responses API output fields
+          - reasoning.encrypted_content
 
 Set a model to ``null`` in a later layer to remove it.
 """
@@ -385,6 +388,8 @@ def get_llm_client(name: str, *, client_type: str | None = None, **overrides) ->
         "allowed_openai_params",
         "additional_drop_params",
         "extra_body",
+        "store",
+        "include",
     ):
         if key in config and key not in overrides:
             params[key] = config[key]