perf(sdk-python): stop re-serializing oversized structured log records

sdk/python/agentfield/logger.py

@@ -213,14 +213,89 @@ def _json_line(record: Dict[str, Any]) -> str:
             record, ensure_ascii=False, separators=(",", ":"), default=str
         )
 
+    @staticmethod
+    def _byte_len(text: str) -> int:
+        """UTF-8 length without materialising a throwaway bytes copy.
+
+        ``str.isascii()`` is an O(1) flag check on CPython's compact-unicode
+        representation, so the common ASCII case never allocates.
+        """
+        return len(text) if text.isascii() else len(text.encode("utf-8"))
+
+    @staticmethod
+    def _json_key(key: Any) -> str:
+        """The string json.dumps() will actually emit for a dict key.
+
+        json coerces non-string keys (``1`` -> ``"1"``, ``True`` -> ``"true"``,
+        ``None`` -> ``"null"``); measuring the raw key instead undercounts by
+        the two quote bytes and can push an elision boundary the wrong way.
+        ``bool`` is checked before ``int`` because ``isinstance(True, int)``.
+        """
+        if isinstance(key, str):
+            return key
+        if isinstance(key, bool):
+            return "true" if key else "false"
+        if key is None:
+            return "null"
+        if isinstance(key, (int, float)):
+            return json.dumps(key)
+        return str(key)
+
+    @classmethod
+    def _shallow_payload_bytes(cls, value: Any) -> int:
+        """Lower bound on the JSON size of one value, without recursing."""
+        if isinstance(value, str):
+            return cls._byte_len(value)
+        if isinstance(value, (bytes, bytearray)):
+            return len(value)
+        return 0
+
+    @classmethod
+    def _certainly_exceeds_budget(cls, record: Dict[str, Any], budget: int) -> bool:
+        """True only when a shallow scan already proves the record is oversized.
+
+        Deliberately shallow: it sums the top-level str/bytes payloads (the
+        record's own values plus one level of ``attributes``) and aborts as
+        soon as the running total passes the budget. Nested containers count
+        as zero, so a False answer means "unknown, serialize and measure".
+        Any exception falls back to that same path — a mirror line is never
+        dropped because the estimator misbehaved.
+        """
+        try:
+            total = 0
+            for key, value in record.items():
+                total += cls._shallow_payload_bytes(value)
+                if total > budget:
+                    return True
+                if key == "attributes" and isinstance(value, dict):
+                    for attribute_value in value.values():
+                        total += cls._shallow_payload_bytes(attribute_value)
+                        if total > budget:
+                            return True
+            return False
+        except Exception:
+            return False
+
+    def _record_size(self, record: Dict[str, Any], attributes_size: int) -> int:
+        """len(_json_line(record)) without re-encoding the attributes payload."""
+        if "attributes" not in record:
+            # {**record, "attributes": {}} would APPEND the key and overcount.
+            return self._byte_len(self._json_line(record))
+        envelope = self._byte_len(self._json_line({**record, "attributes": {}}))
+        return envelope - 2 + attributes_size  # -2 drops the "{}" placeholder
+
     def _bounded_mirror_line(self, record: Dict[str, Any]) -> str:
         """Serialize a valid JSON view while leaving the dispatched record untouched."""
         from .node_logs import max_line_bytes
 
         budget = max_line_bytes()
-        line = self._json_line(record)
-        if len(line.encode("utf-8")) <= budget:
-            return line
+        line: Optional[str] = None
+        record_size: Optional[int] = None
+        if not self._certainly_exceeds_budget(record, budget):
+            line = self._json_line(record)
+            record_size = self._byte_len(line)
+            if record_size <= budget:
+                return line
 
         # Shallow copies are enough: attribute values are replaced, never
         # mutated, and a deepcopy of a multi-megabyte (or non-copyable)
@@ -232,8 +307,8 @@ def _bounded_mirror_line(self, record: Dict[str, Any]) -> str:
             view["attributes"] = attributes
             encoded_sizes = [
                 (
-                    len(self._json_line(value).encode("utf-8")),
-                    len(self._json_line(key).encode("utf-8")),
+                    self._byte_len(self._json_line(value)),
+                    self._byte_len(self._json_line(self._json_key(key))),
                     key,
                 )
                 for key, value in attributes.items()
@@ -244,13 +319,15 @@ def _bounded_mirror_line(self, record: Dict[str, Any]) -> str:
             )
             if encoded_sizes:
                 attributes_size += len(encoded_sizes) - 1
+            if record_size is None:
+                record_size = self._record_size(record, attributes_size)
 
-            estimated_size = len(line.encode("utf-8"))
+            estimated_size = record_size
             for size, _key_size, key in sorted(
                 encoded_sizes, key=lambda item: item[0], reverse=True
             ):
                 marker = f"<{size} bytes elided>"
-                marker_size = len(self._json_line(marker).encode("utf-8"))
+                marker_size = self._byte_len(self._json_line(marker))
                 attributes[key] = marker
                 estimated_size -= size - marker_size
                 if estimated_size <= budget:
@@ -259,41 +336,39 @@ def _bounded_mirror_line(self, record: Dict[str, Any]) -> str:
             if estimated_size > budget:
                 view["attributes"] = {"_elided": f"<{attributes_size} bytes elided>"}
         else:
-            size = len(self._json_line(attributes).encode("utf-8"))
+            size = self._byte_len(self._json_line(attributes))
+            if record_size is None:
+                record_size = self._record_size(record, size)
             view["attributes"] = f"<{size} bytes elided>"
 
         line = self._json_line(view)
-        if len(line.encode("utf-8")) <= budget:
+        if self._byte_len(line) <= budget:
             return line
 
         message = str(view.get("message", ""))
-        message_bytes = message.encode("utf-8")
+        message_size = self._byte_len(message)
         low, high = 0, len(message)
         while low <= high:
             keep = (low + high) // 2
-            elided = len(message_bytes) - len(message[:keep].encode("utf-8"))
+            elided = message_size - self._byte_len(message[:keep])
             view["message"] = f"{message[:keep]}…[{elided} bytes elided]"
             line = self._json_line(view)
-            if len(line.encode("utf-8")) <= budget:
+            if self._byte_len(line) <= budget:
                 low = keep + 1
             else:
                 high = keep - 1
         if high >= 0:
             kept = message[:high]
-            elided = len(message_bytes) - len(kept.encode("utf-8"))
+            elided = message_size - self._byte_len(kept)
             view["message"] = f"{kept}…[{elided} bytes elided]"
             line = self._json_line(view)
-            if len(line.encode("utf-8")) <= budget:
+            if self._byte_len(line) <= budget:
                 return line
 
-        original_size = len(self._json_line(record).encode("utf-8"))
+        original_size = record_size
 
         def bounded_scalar(value: Any) -> Any:
-            return (
-                value
-                if len(self._json_line(value).encode("utf-8")) <= 32
-                else "<elided>"
-            )
+            return value if self._byte_len(self._json_line(value)) <= 32 else "<elided>"
 
         minimal = {
             "timestamp": bounded_scalar(record.get("ts")),
@@ -303,7 +378,7 @@ def bounded_scalar(value: Any) -> Any:
         }
         line = self._json_line(minimal)
         # max_line_bytes() has a 256-byte floor; this fixed envelope is smaller.
-        assert len(line.encode("utf-8")) <= budget
+        assert self._byte_len(line) <= budget
         return line
 
     def _dispatch_to_cp(self, record: Dict[str, Any]) -> None:

sdk/python/tests/test_logger.py

@@ -363,6 +363,163 @@ def test_structured_mirror_many_large_attributes_stays_fast(monkeypatch):
     assert elapsed < 0.2
 
 
+@pytest.mark.unit
+def test_structured_mirror_serializes_large_string_payload_at_most_once(monkeypatch):
+    budget = 16384
+    monkeypatch.setenv("AGENTFIELD_LOG_MAX_LINE_BYTES", str(budget))
+    stream = io.StringIO()
+    monkeypatch.setattr(logger_module.sys, "stdout", stream)
+    calls = []
+    real_dumps = logger_module.json.dumps
+
+    def counting_dumps(obj, **kwargs):
+        result = real_dumps(obj, **kwargs)
+        # Classify by OUTPUT size: the redundant call this guards against
+        # serializes the whole record, not the oversized string itself.
+        calls.append(len(result))
+        return result
+
+    monkeypatch.setattr(logger_module.json, "dumps", counting_dumps)
+    record = {
+        "message": "bounded",
+        "attributes": {"binary": b"x", "payload": "x" * 5_000_000},
+    }
+
+    AgentFieldLogger("large-string-once")._emit_structured_record(record)
+
+    line = stream.getvalue().rstrip("\n")
+    json.loads(line)
+    assert len(line.encode("utf-8")) <= budget
+    assert sum(1 for size in calls if size >= 1_000_000) == 1
+    assert record["attributes"]["payload"] == "x" * 5_000_000
+
+
+@pytest.mark.unit
+def test_structured_mirror_fitting_nested_record_is_not_walked_recursively(
+    monkeypatch,
+):
+    walked = []
+
+    class _TripwireStr(str):
+        """A nested string a recursive precheck would touch; json.dumps does not."""
+
+        # Recording rather than raising is deliberate: _certainly_exceeds_budget
+        # swallows every exception and falls back to the full-dumps path, so a
+        # raising tripwire would be invisible to the caller.
+        def isascii(self):
+            walked.append("isascii")
+            return str.isascii(self)
+
+        def __len__(self):
+            walked.append("len")
+            return str.__len__(self)
+
+    monkeypatch.setenv("AGENTFIELD_LOG_MAX_LINE_BYTES", "16384")
+    stream = io.StringIO()
+    monkeypatch.setattr(logger_module.sys, "stdout", stream)
+    record = {
+        "message": "bounded",
+        "attributes": {
+            "rows": [{"i": index, "n": _TripwireStr("x")} for index in range(700)]
+        },
+    }
+    expected = json.dumps(
+        record, ensure_ascii=False, separators=(",", ":"), default=str
+    )
+
+    AgentFieldLogger("fitting-nested")._emit_structured_record(record)
+
+    line = stream.getvalue().rstrip("\n")
+    mirrored = json.loads(line)
+    assert line == expected
+    assert walked == []
+    assert len(mirrored["attributes"]["rows"]) == 700
+
+
+@pytest.mark.unit
+def test_structured_mirror_non_string_attribute_keys_have_exact_sizes(monkeypatch):
+    monkeypatch.setenv("AGENTFIELD_LOG_MAX_LINE_BYTES", "512")
+    stream = io.StringIO()
+    monkeypatch.setattr(logger_module.sys, "stdout", stream)
+    attributes = {0: "x" * 4000, True: "y", None: "z"}
+    logger = AgentFieldLogger("non-string-keys")
+
+    logger._emit_structured_record({"message": "kept", "attributes": attributes})
+    line = stream.getvalue().rstrip("\n")
+    assert json.loads(line)["attributes"]["0"] == "<4002 bytes elided>"
+
+    stream.seek(0)
+    stream.truncate(0)
+    logger._emit_structured_record({"message": "z" * 100_000, "attributes": attributes})
+    line = stream.getvalue().rstrip("\n")
+    attributes_size = len(
+        json.dumps(attributes, ensure_ascii=False, separators=(",", ":"), default=str)
+    )
+    assert json.loads(line)["attributes"]["_elided"] == (
+        f"<{attributes_size} bytes elided>"
+    )
+
+
+@pytest.mark.unit
+def test_structured_mirror_without_attributes_reports_exact_original_size(monkeypatch):
+    monkeypatch.setenv("AGENTFIELD_LOG_MAX_LINE_BYTES", "256")
+    stream = io.StringIO()
+    monkeypatch.setattr(logger_module.sys, "stdout", stream)
+    record = {
+        "ts": "t" * 1000,
+        "level": "info",
+        "message": "z" * 100_000,
+    }
+    expected_size = len(
+        json.dumps(record, ensure_ascii=False, separators=(",", ":"), default=str)
+    )
+
+    AgentFieldLogger("missing-attributes")._emit_structured_record(record)
+
+    line = stream.getvalue().rstrip("\n")
+    assert json.loads(line)["message"] == f"<record elided: {expected_size} bytes>"
+
+
+@pytest.mark.unit
+def test_structured_mirror_precheck_exception_falls_back_to_serialization(monkeypatch):
+    class _TripwireStr(str):
+        def isascii(self):
+            raise RecursionError("precheck measurement failed")
+
+        def __len__(self):
+            raise RecursionError("precheck measurement failed")
+
+    budget = 512
+    monkeypatch.setenv("AGENTFIELD_LOG_MAX_LINE_BYTES", str(budget))
+    stream = io.StringIO()
+    monkeypatch.setattr(logger_module.sys, "stdout", stream)
+    record = {"message": "kept", "attributes": {"value": _TripwireStr("safe")}}
+
+    AgentFieldLogger("precheck-exception")._emit_structured_record(record)
+
+    line = stream.getvalue().rstrip("\n")
+    assert json.loads(line)["attributes"]["value"] == "safe"
+    assert len(line.encode("utf-8")) <= budget
+
+
+@pytest.mark.unit
+@pytest.mark.parametrize(
+    ("key", "expected"),
+    [("key", "key"), (False, "false"), (None, "null"), (1.5, "1.5")],
+)
+def test_json_key_matches_json_dict_key_coercion(key, expected):
+    assert AgentFieldLogger._json_key(key) == expected
+
+
+@pytest.mark.unit
+def test_json_key_falls_back_to_string_for_unknown_key_type():
+    class Key:
+        def __str__(self):
+            return "custom"
+
+    assert AgentFieldLogger._json_key(Key()) == "custom"
+
+
 @pytest.mark.unit
 def test_public_structured_logger_emits_only_bounded_json_lines(monkeypatch):
     cap = 512