fix(sessions): validate imported values before persistence

Session import used shallow type casts at the persistence boundary:
float() accepted inf/NaN (a JSON 1e309), int() accepted integers
SQLite cannot bind (outside signed 64-bit range), model_config was
only shape-checked after its intentional decode, and a literal string
carrying the internal \\x00json: content marker was decoded as
structured content on read. Bad values reached sqlite3 binding or
persisted as inf, breaking session APIs with HTTP 500.

- reject non-finite floats in session float/timestamp columns, message
  timestamps, and scalar message content, with the field path in the error
- re-check every intentionally-decoded JSON-backed field (model_config,
  reasoning_details, codex_* items) for non-finite values
- bound int columns, token_count, and int scalar content to SQLite's
  signed 64-bit range; coerce-with-default at the bind site as a backstop
- escape literal \\x00json:-prefixed strings via the JSON wrapper instead
  of storing them raw; decoding already fails closed on malformed marker rows
- all rejection stays inside the existing atomic validate-then-write
  400 contract; valid exports round-trip unchanged

Fixes #63954

hermes_state_messages.py

@@ -138,10 +138,18 @@ class SessionMessagesMixin:
         """Serialize list/dict content (multimodal parts) as a sentinel-prefixed JSON string (sqlite3 binds
         only scalars). Lone UTF-16 surrogates (unsanitized web-scraped tool results) are scrubbed here: left
         raw, sqlite3 raises UnicodeEncodeError and the session silently stops persisting. Pairs with
-        :meth:`_decode_content`."""
+        :meth:`_decode_content`.
+
+        A literal string carrying the reserved ``\\x00json:`` prefix is indistinguishable from
+        stored structured content, so it is JSON-string-wrapped (decoded back verbatim on read)
+        instead of written raw — a crafted import can't smuggle the marker in and have the read
+        path decode attacker-chosen structure."""
         if isinstance(content, str):
-            return _sanitize_surrogates(content)
-        if content is None or isinstance(content, (bytes, int, float)):
+            content = _sanitize_surrogates(content)
+            if not content.startswith(cls._CONTENT_JSON_PREFIX):
+                return content
+            # Fall through: escape the reserved prefix by double-encoding the literal.
+        elif content is None or isinstance(content, (bytes, int, float)):
             return content
         try:
             return cls._CONTENT_JSON_PREFIX + json.dumps(content)  # ensure_ascii escapes surrogates: bindable
@@ -150,7 +158,8 @@ class SessionMessagesMixin:
 
     @classmethod
     def _decode_content(cls, content: Any) -> Any:
-        """Reverse :meth:`_encode_content`; returns scalars unchanged."""
+        """Reverse :meth:`_encode_content`; returns scalars unchanged. Fail closed: a reserved-prefix
+        string that isn't valid JSON (a legacy/malformed marker row) returns the raw string."""
         if isinstance(content, str) and content.startswith(cls._CONTENT_JSON_PREFIX):
             return _json_or(content[len(cls._CONTENT_JSON_PREFIX):], content,
                 "Failed to decode JSON-encoded message content; returning raw string")

hermes_state_portability.py

@@ -6,6 +6,7 @@ Must never import hermes_state (cycle); shared constants live in hermes_state_co
 
 import logging
 import json
+import math
 import time
 from collections import Counter
 from datetime import datetime, timezone
@@ -436,12 +437,46 @@ class SessionPortabilityMixin:
             raise ValueError(f"{field} must be an integer") from exc
 
     @staticmethod
-    def _import_json_object_or_none(value: Any, field: str) -> Optional[str]:
+    def _import_non_finite_path(value: Any, path: str) -> Optional[str]:
+        """First ``path`` of a NaN/inf number anywhere in decoded JSON data, else None.
+
+        Iterative walk (a recursive visitor would overflow on crafted nesting); cycles
+        can't arrive through ``json.loads`` but guard anyway.
+        """
+        pending = [(value, path)]
+        while pending:
+            current, current_path = pending.pop()
+            if isinstance(current, float) and not math.isfinite(current):
+                return current_path
+            if isinstance(current, dict):
+                pending.extend((child, f"{current_path}.{key}") for key, child in current.items())
+            elif isinstance(current, list):
+                pending.extend((child, f"{current_path}[{key}]") for key, child in enumerate(current))
+        return None
+
+    @classmethod
+    def _import_load_json(cls, value: str, path: str) -> Any:
+        """``json.loads`` that refuses NaN/Infinity anywhere in the decoded value.
+
+        A JSON exponent like ``1e309`` decodes to ``float("inf")``, which strict
+        serializers reject and cost analytics mis-handle. Callers that intentionally
+        decode a JSON-backed column (model_config, reasoning items) re-check through
+        here so non-finite values can't reappear after the outer payload was inspected.
+        """
+        parsed = json.loads(value)  # NaN/Infinity are accepted syntax; catch them after decode
+        non_finite = cls._import_non_finite_path(parsed, path)
+        if non_finite is not None:
+            raise ValueError(f"non-finite number at {non_finite}")
+        return parsed
+
+    @classmethod
+    def _import_json_object_or_none(cls, value: Any, field: str) -> Optional[str]:
         if value is None:
             return None
+        path = f"session.{field}"
         if isinstance(value, str):
             try:
-                parsed = json.loads(value)
+                parsed = cls._import_load_json(value, path)
             except json.JSONDecodeError as exc:
                 raise ValueError(f"{field} must be valid JSON") from exc
             if not isinstance(parsed, dict):
@@ -449,6 +484,8 @@ class SessionPortabilityMixin:
             return value
         if not isinstance(value, dict):
             raise ValueError(f"{field} must be a JSON object")
+        if cls._import_non_finite_path(value, path) is not None:
+            raise ValueError(f"non-finite number at {path}")
         try:
             return json.dumps(value)
         except (TypeError, ValueError) as exc:
@@ -462,6 +499,83 @@ class SessionPortabilityMixin:
         except (TypeError, ValueError):
             return default
 
+    # SQLite binds Python ints only inside the signed 64-bit range; anything wider
+    # raises OverflowError at INSERT time (an unhandled 500, not a 400).
+    _SQLITE_INT_MIN = -(1 << 63)
+    _SQLITE_INT_MAX = (1 << 63) - 1
+
+    @classmethod
+    def _import_int_or_bounded(cls, value: Any, field: str, default: int) -> int:
+        """Import an int column: SQL-int-range or *default* (never fails the import)."""
+        number = cls._coerce_or(value, int, default)
+        if number is None:
+            return default
+        if isinstance(number, bool) or not (cls._SQLITE_INT_MIN <= number <= cls._SQLITE_INT_MAX):
+            return default
+        return number
+
+    def _normalize_import_message(self, message: Dict[str, Any], message_index: int) -> Dict[str, Any]:
+        """Type-check one payload message; raises ValueError naming the offending field."""
+        clean_message = dict(message)
+        role = clean_message.get("role")
+        if not isinstance(role, str) or not role:
+            raise ValueError(f"messages[{message_index}].role must be a non-empty string")
+        for field in _IMPORT_MESSAGE_TEXT_FIELDS:
+            clean_message[field] = self._import_text_or_none(clean_message.get(field), field)
+        token_count = self._import_int_or_none(clean_message.get("token_count"), "token_count")
+        if token_count is not None and not (self._SQLITE_INT_MIN <= token_count <= self._SQLITE_INT_MAX):
+            raise ValueError("token_count is outside SQLite's integer range")
+        clean_message["token_count"] = token_count
+        message_path = f"session.messages[{message_index}]"
+        timestamp = clean_message.get("timestamp")
+        if timestamp is not None and coerce_epoch(timestamp, field="messages[].timestamp") is None:
+            raise ValueError(f"{message_path}.timestamp must be a supported Unix timestamp")
+        self._validate_import_message_content(clean_message.get("content"), message_path)
+        for field in _IMPORT_MESSAGE_JSON_FIELDS:
+            json_value = clean_message.get(field)
+            if isinstance(json_value, str):
+                # Malformed JSON text persists verbatim (readers fall back via _json_or);
+                # only a PARSED non-finite number is rejected — it would break strict
+                # JSON serializers once decoded back out of the row.
+                try:
+                    self._import_load_json(json_value, f"{message_path}.{field}")
+                except json.JSONDecodeError:
+                    pass
+                except ValueError:
+                    raise
+            elif json_value is not None:
+                non_finite = self._import_non_finite_path(json_value, f"{message_path}.{field}")
+                if non_finite is not None:
+                    raise ValueError(f"non-finite number at {non_finite}")
+        return clean_message
+
+    @classmethod
+    def _validate_import_message_content(cls, content: Any, message_path: str) -> None:
+        """Scalar message content must be bindable and later serializable."""
+        if isinstance(content, int) and not isinstance(content, bool) and not (
+                cls._SQLITE_INT_MIN <= content <= cls._SQLITE_INT_MAX):
+            # SQLite can't bind a wider int: this reached INSERT and raised OverflowError.
+            raise ValueError(f"{message_path}.content is outside SQLite's integer range")
+        if isinstance(content, float) and not math.isfinite(content):
+            raise ValueError(f"non-finite number at {message_path}.content")
+        if isinstance(content, (bytes, bytearray)):
+            raise ValueError(f"{message_path}.content must be a string, list, object, finite number, or null")
+
+    @classmethod
+    def _import_float_or_finite(cls, value: Any, field: str) -> Optional[float]:
+        """Import a float column: finite float or None (never fails the import).
+
+        A JSON exponent like ``1e309`` decodes to ``float("inf")``; persisted, it later
+        breaks strict JSON serialization in the session APIs. Missing stays missing.
+        """
+        if value is None:
+            return None
+        try:
+            number = float(value)
+        except (TypeError, ValueError, OverflowError):
+            return None
+        return number if math.isfinite(number) else None
+
     def _normalize_import_session(self, raw: Dict[str, Any], session_id: str, messages: list) -> Dict[str, Any]:
         """Type-check one payload session + its messages; raises ValueError."""
         clean_session = dict(raw)
@@ -469,16 +583,21 @@ class SessionPortabilityMixin:
         clean_session["model_config"] = self._import_json_object_or_none(clean_session.get("model_config"), "model_config")
         for field in ("parent_session_id", *_IMPORT_SESSION_TEXT_FIELDS):
             clean_session[field] = self._import_text_or_none(clean_session.get(field), field)
+        started_at = clean_session.get("started_at")
+        if started_at is not None and coerce_epoch(started_at, field="started_at") is None:
+            raise ValueError("session.started_at must be a supported Unix timestamp")
+        for field in _IMPORT_FLOAT_COLS:
+            value = clean_session.get(field)
+            if isinstance(value, float) and not math.isfinite(value):
+                raise ValueError(f"non-finite number at session.{field}")
+        for field in _IMPORT_INT_COLS:
+            value = clean_session.get(field)
+            if isinstance(value, int) and not isinstance(value, bool) and not (
+                    self._SQLITE_INT_MIN <= value <= self._SQLITE_INT_MAX):
+                raise ValueError(f"{field} is outside SQLite's integer range")
         clean_messages: List[Dict[str, Any]] = []
         for message_index, message in enumerate(messages):
-            clean_message = dict(message)
-            role = clean_message.get("role")
-            if not isinstance(role, str) or not role:
-                raise ValueError(f"messages[{message_index}].role must be a non-empty string")
-            for field in _IMPORT_MESSAGE_TEXT_FIELDS:
-                clean_message[field] = self._import_text_or_none(clean_message.get(field), field)
-            clean_message["token_count"] = self._import_int_or_none(clean_message.get("token_count"), "token_count")
-            clean_messages.append(clean_message)
+            clean_messages.append(self._normalize_import_message(message, message_index))
         return {"session": clean_session, "messages": clean_messages}
 
     def _validate_import_payload(self, sessions: List[Dict[str, Any]]) -> tuple:
@@ -545,8 +664,8 @@ class SessionPortabilityMixin:
             "started_at": time.time() if started_at is None else started_at,
             **{col: 1 if self._coerce_or(raw.get(col), int, 0) else 0 for col in _IMPORT_FLAG_COLS},
             **{col: raw.get(col) for col in _IMPORT_PASSTHROUGH_COLS},
-            **{col: self._coerce_or(raw.get(col), float, None) for col in _IMPORT_FLOAT_COLS},
-            **{col: self._coerce_or(raw.get(col), int, 0) for col in _IMPORT_INT_COLS},
+            **{col: self._import_float_or_finite(raw.get(col), col) for col in _IMPORT_FLOAT_COLS},
+            **{col: self._import_int_or_bounded(raw.get(col), col, 0) for col in _IMPORT_INT_COLS},
         }
         params["auto_archived"] &= params["archived"]  # only the sweep sets it, and only on rows it archives
         conn.execute(_IMPORT_SESSION_INSERT_SQL, params)

tests/hermes_state/test_session_import_value_validation.py

@@ -0,0 +1,135 @@
+"""Session-import boundary validation for values later read by session APIs.
+
+``import_sessions`` used shallow type casts at the persistence boundary: ``float()`` accepted
+``inf``/``NaN`` (a JSON ``1e309``), ``int()`` accepted integers SQLite cannot bind (outside the
+signed 64-bit range), ``model_config`` was only checked for being a JSON *object* after its
+intentional decode (non-finite values inside it persisted verbatim), and a literal string
+carrying the internal ``\\x00json:`` content marker was decoded as structured content on read.
+Bad values therefore reached sqlite3 binding or persisted as ``inf``, breaking session APIs
+with HTTP 500. Regression for #63954 (validation shape reused from PR #63956 by @Jupiter363).
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from hermes_state import SessionDB
+
+
+@pytest.fixture()
+def db(tmp_path):
+    d = SessionDB(db_path=tmp_path / "state.db")
+    yield d
+    d.close()
+
+
+def _payload(session_id: str, **overrides) -> dict:
+    session = {"id": session_id, "messages": [{"role": "user", "content": "hello"}]}
+    session.update(overrides)
+    return session
+
+
+def _rejected(db: SessionDB, session: dict) -> str:
+    """Import must reject atomically (structured 400 contract) and name the field."""
+    result = db.import_sessions([session])
+    assert result["ok"] is False, "unsafe import was accepted"
+    assert result["imported"] == 0
+    assert db.get_session(session["id"]) is None, "a rejected import still wrote a row"
+    error = result["errors"][0]["error"]
+    assert error, "rejection carried no error message"
+    return error
+
+
+def test_import_rejects_non_finite_session_floats(db):
+    for field in ("started_at", "ended_at", "estimated_cost_usd", "actual_cost_usd"):
+        # 1e309 decodes to float('inf'); NaN arrives the same way (json NaN syntax).
+        for value in (1e309, float("nan")):
+            error = _rejected(db, _payload(f"bad-{field}-{value}", **{field: value}))
+            assert "non-finite" in error or "timestamp" in error, (field, error)
+
+
+def test_import_rejects_non_finite_nested_model_config(db):
+    error = _rejected(db, _payload("nested", model_config='{"temperature": 1e309}'))
+    assert "non-finite number at session.model_config" in error
+    # Also as a decoded dict, not only as a JSON string.
+    error = _rejected(db, _payload("nested-dict", model_config={"temperature": float("inf")}))
+    assert "non-finite" in error
+
+
+def test_import_rejects_non_finite_reasoning_json(db):
+    session = _payload(
+        "reasoning",
+        messages=[{"role": "assistant", "content": "hi",
+                   "reasoning_details": '[{"temperature": 1e309}]'}],
+    )
+    error = _rejected(db, session)
+    assert "non-finite number at session.messages[0].reasoning_details" in error
+
+
+def test_import_rejects_int_columns_outside_sqlite_range(db):
+    huge = 1 << 63
+    for field in ("input_tokens", "output_tokens", "api_call_count", "cache_read_tokens",
+                  "cache_write_tokens", "reasoning_tokens"):
+        assert "outside SQLite's integer range" in _rejected(db, _payload(f"huge-{field}", **{field: huge})), field
+
+
+def test_import_rejects_oversized_scalar_message_content(db):
+    error = _rejected(db, _payload("huge-content", messages=[
+        {"role": "user", "content": 1 << 63}]))
+    assert "content is outside SQLite's integer range" in error
+    error = _rejected(db, _payload("inf-content", messages=[
+        {"role": "user", "content": float("inf")}]))
+    assert "non-finite number at session.messages[0].content" in error
+
+
+def test_import_rejects_oversized_token_count_and_bad_message_timestamp(db):
+    assert "token_count" in _rejected(db, _payload("huge-tokens", messages=[
+        {"role": "user", "content": "hi", "token_count": 1 << 63}]))
+    assert "timestamp" in _rejected(db, _payload("bad-ts", messages=[
+        {"role": "user", "content": "hi", "timestamp": 1e309}]))
+
+
+def test_reserved_content_prefix_literal_round_trips_verbatim(db):
+    """A literal string imitating the internal ``\\x00json:`` marker must NOT be decoded
+    as structured content on read — it comes back as the exact literal string."""
+    marker = SessionDB._CONTENT_JSON_PREFIX
+    literal = marker + "not actually json"
+    db.import_sessions([_payload("marker", messages=[{"role": "user", "content": literal}])])
+    assert db.get_messages("marker")[0]["content"] == literal
+    # And the encoded form of a marker-prefixed string still decodes to the literal.
+    encoded = SessionDB._encode_content(literal)
+    assert encoded.startswith(marker)
+    assert SessionDB._decode_content(encoded) == literal
+
+
+def test_decode_content_fails_closed_on_malformed_marker_rows(db):
+    malformed = SessionDB._CONTENT_JSON_PREFIX + "[{not json"
+    assert SessionDB._decode_content(malformed) == malformed
+
+
+def test_normal_export_round_trips_unchanged(db):
+    """The primary compatibility contract: an ordinary session exports and re-imports
+    with content, timestamps and counters intact."""
+    db.create_session(session_id="rt-src", source="cli", model="test-model")
+    db.append_message("rt-src", "user", "hello world", token_count=5)
+    db.append_message("rt-src", "assistant", [{"type": "text", "text": "hi"}],
+                      token_count=7, finish_reason="stop",
+                      reasoning_details=[{"type": "summary", "text": "because"}])
+    db.end_session("rt-src", "done")
+    payload = json.loads(json.dumps(db.export_session("rt-src")))
+
+    result = db.import_sessions([dict(payload, id="rt-copy")])
+    assert result["ok"] is True and result["imported"] == 1
+
+    restored = db.export_session("rt-copy")
+    for message in (payload["messages"], restored["messages"]):
+        message.sort(key=lambda m: m.get("timestamp") or 0)
+    for original, copy in zip(payload["messages"], restored["messages"]):
+        assert copy["content"] == original["content"]
+        assert copy.get("token_count") == original.get("token_count")
+        assert copy["role"] == original["role"]
+    assert restored["message_count"] == payload["message_count"]
+    assert restored["tool_call_count"] == payload["tool_call_count"]
+    assert restored["ended_at"] is not None