diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/README.md b/packages/aws-durable-execution-sdk-python-telemetry-exporter/README.md
index eb6bd9ed1..056708be9 100644
--- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/README.md
+++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/README.md
@@ -47,8 +47,8 @@ state is per invocation. Its type is `DurableExecutionTelemetryPluginFactory`, e
 the package root for annotating a value you hold.
 
 With no exporter configured, records are written to the function's own
-CloudWatch log group as single JSON lines (the `LambdaLogExporter` default),
-carrying the name-keyed `operationsByName` summary. The `S3Exporter` writes the
+CloudWatch log group as single JSON lines (the `LambdaLogExporter` default, see
+below), carrying the name-keyed `operationsByName` summary. The `S3Exporter` writes the
 lossless per-occurrence `operations` array, one object per execution
 (upsert-by-execution-name, so re-emission overwrites rather than appends).
 
@@ -80,6 +80,38 @@ the Lambda runtime's boto3 otherwise; none adds a required dependency.
 size limit; omitting it keeps the default. `HttpExporter` and `FileExporter`
 have no default and do not truncate unless a limit is set.
 
+### LambdaLogExporter
+
+Prints one line per record to stdout, which Lambda sends to the function's own
+log group. The line is the Lambda JSON log envelope, with keys in this order:
+
+```json
+{"timestamp":"2026-01-01T00:00:01.234Z","level":"INFO","message":"<compact record JSON>"}
+```
+
+`timestamp` is UTC with millisecond precision and a `Z` suffix. `message` is
+the compact record JSON as a string; parse it to get the record. The exporter
+writes the envelope itself because Lambda captures Python `print` output as
+plain text, in both the plain text and the JSON log format, and Lambda does not
+double-encode output that is already JSON (see
+[Log and monitor Python Lambda functions](https://docs.aws.amazon.com/lambda/latest/dg/python-logging.html)
+and [Lambda log formats](https://docs.aws.amazon.com/lambda/latest/dg/monitoring-cloudwatchlogs-logformat.html)).
+
+`max_record_size_bytes` (default 256,000 bytes) limits the printed line: the
+envelope, including the JSON escaping of `message`, in UTF-8 bytes without the
+trailing newline. Truncation measures that line in one pass. A CloudWatch Logs
+event can be up to 1 MB
+([PutLogEvents](https://docs.aws.amazon.com/AmazonCloudWatchLogs/latest/APIReference/API_PutLogEvents.html)).
+U+2028 and U+2029 are written as `\u2028` and `\u2029`, so a tool that treats
+them as line breaks still sees one line per record. A lone surrogate (a string
+character with no UTF-8 encoding) is written as its `\uXXXX` escape, so
+printing it never fails. The limit counts these escapes; other exporters'
+limits count the characters as they write them.
+
+The lines have `level` `INFO`. With the JSON log format and an application log
+level of `WARN` or above, Lambda drops them. With the Text log format, Lambda
+does not filter them.
+
 ### CloudWatchLogsExporter
 
 Writes one `PutLogEvents` event per record to `log_group_name`, in a stream
@@ -95,6 +127,13 @@ CloudWatchLogsExporter(log_group_name="/custom/durable-execution-telemetry")
 `PutItem` keyed by `partition_key` (default `pk`) = `executionArn`. With the
 default `sort_key="sk"` (= `emittedAt`) every export adds an item; pass
 `sort_key=None` for a key-only table that upserts. IAM: `dynamodb:PutItem`.
+A number DynamoDB cannot store (more than 38 significant digits, a magnitude
+outside 1E-130 to 9.99...E+125, NaN or Infinity) is written as a string
+attribute holding its JSON text, so it never drops the item. The same field can
+therefore be a number (`N`) in one item and a string (`S`) in another; readers
+that filter or compare on the attribute type should allow for both. An integer
+with more than 38 digits but at most 38 significant digits, such as `10**38`,
+is stored as a number.
 
 ```python
 DynamoDBExporter(table_name="durable-execution-telemetry")
@@ -241,6 +280,77 @@ Behavior is validated cross-SDK by the `telemetry` conformance suite
 > and any earlier flushes. Budget for that finite backlog, especially when an
 > exporter makes a network call per record.
 
+## Error names and causes
+
+Every error in a record (the execution `error`, each `operations[].error`, and
+each `operationsByName.<name>.error`) has the shape
+`{"name": ..., "message": ..., "cause": ...}`, with keys in that order. `name`
+is the SDK's error type, unchanged, and `message` is the error message. `cause`
+has the same shape, recursively, and is present only when the SDK provides a
+cause.
+
+In Python, the plugin hooks carry only an `ErrorObject` (type, message, data,
+stack trace) with no cause, so no error in a Python record has a `cause`: not
+the execution `error`, not an operation error. The exporter does not build a
+cause from other operations or from `ErrorData`. `include_errors=False`
+removes operation errors.
+
+Every `name` and `message` is capped at 8,192 characters, counted as Unicode
+code points (a Python `str` is a sequence of code points, so a cut never splits
+a character). A longer value keeps its first 8,192 code points followed by `…`.
+
+When a record is over its exporter's size limit, truncation first removes every
+`cause` and sets `droppedCauses: true`, then runs its usual steps (operation
+results, then whole operations, then input, then output). Python records have
+no cause, so the first step never changes them. With the name and message
+caps, a record whose operations, input and output are dropped fits any limit of
+64 KB or more, assuming ARN fields of normal length (a few hundred bytes, as in
+real Lambda ARNs) and an error `name` of normal length (a type name such as
+`module.QualName`). Without the `name` assumption, the worst case (a name and
+message of 8,192 control characters, each escaped as `\u00XX`) fits any limit
+of 128 KB (131,072 bytes) or more.
+
+Names are each language's own and differ across SDKs. For example, a step that
+throws `PaymentError` can be reported as `StepError` in JS,
+`aws_durable_execution_sdk_python.exceptions.StepError` or `PaymentError` in
+Python, and `StepFailedException` or `PaymentError` in Java. Query error names
+per SDK, or query on `message` and the `cause` chain.
+
+## Operation results
+
+An opted-in operation `result` (`OperationOverride(result=...)`) is passed to
+the transform as a plain JSON value. When the checkpointed string is the
+default serializer's (`ExtendedTypeSerDes`) envelope, a JSON object whose keys
+are exactly `t` and `v` with a string `t`, it is decoded with that serializer
+and converted to plain JSON:
+
+- `int`: unchanged when its absolute value is at most `2**53 - 1`
+  (9007199254740991), the largest integer a JavaScript reader parses exactly.
+  Otherwise `str(value)`, for example `"12345678901234567"`.
+- `Decimal`: an integer when it is integral and its absolute value is at most
+  `2**53 - 1`. Otherwise a float when the float round-trips exactly and is
+  within DynamoDB's number range (magnitude 1E-130 to 9.99...E+125). Otherwise `str(value)`, for example `"9007199254740993"`,
+  `"12345678901234567.89"` or `"1E+299"`. NaN and Infinity become `null`.
+  Negative zero (`Decimal("-0")`, `Decimal("-0.0")`) is integral, so it becomes
+  `0` and loses its sign.
+- `datetime` and `date`: `isoformat()`. `UUID`: string. `bytes`: base64.
+  `tuple`: list. `BatchResult`: `to_dict()`.
+
+Any other result is parsed as JSON, or kept as the raw string when it is not
+JSON. In parsed JSON, NaN and Infinity become `null`, and an integer whose
+absolute value is greater than `2**53 - 1` becomes its decimal string, as
+above; nothing else changes. A
+result that fails to decode keeps its parsed or raw value, and the rest of the
+record is still emitted.
+
+The hooks do not say which serializer wrote a result, so the envelope is
+recognized by shape alone. A result written by `JsonSerDes` or a custom
+`SerDes` that has this exact shape is decoded too: `{"t": "s", "v": "x"}` is
+reported as `"x"`. Only the core serializer's known tags are decoded, and the
+core may coerce `v` while decoding: `{"t": "b", "v": "yes"}` is reported as
+`true`. An unknown tag, such as `{"t": "temperature", "v": 21.5}`, fails to
+decode and passes through unchanged.
+
 ## Requirements
 
 - `aws-durable-execution-sdk-python` with the plugin invocation hooks that
diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/_export_scheduler.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/_export_scheduler.py
index 37a7c2074..b798d8c78 100644
--- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/_export_scheduler.py
+++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/_export_scheduler.py
@@ -619,8 +619,13 @@ def _run_loop(self) -> None:
     def _export(self, record: dict[str, Any]) -> None:
         for exporter in self._exporters:
             try:
+                # ``measure`` is optional: an exporter that writes its rendered
+                # value as something other than compact JSON supplies it.
                 shaped = truncate_record(
-                    record, exporter.max_record_size_bytes, exporter.render
+                    record,
+                    exporter.max_record_size_bytes,
+                    exporter.render,
+                    getattr(exporter, "measure", None),
                 )
                 exporter.export(shaped)
             except BaseException as exc:  # noqa: BLE001 - one exporter must not break others
diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/exporters/dynamodb_exporter.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/exporters/dynamodb_exporter.py
index 1b3c9fe13..aacc7b3bb 100644
--- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/exporters/dynamodb_exporter.py
+++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/exporters/dynamodb_exporter.py
@@ -5,8 +5,8 @@
 
 from __future__ import annotations
 
+import decimal
 import json
-from decimal import Decimal
 from typing import Any
 
 from aws_durable_execution_sdk_python_telemetry_exporter.exporters._common import (
@@ -20,6 +20,13 @@
 class DynamoDBExporter:
     """Writes ``operationsByName`` records to a DynamoDB table with PutItem.
 
+    Numbers DynamoDB cannot store (more than 38 significant digits, a
+    magnitude outside 1E-130 to 9.99...E+125, NaN or Infinity) are written as
+    their JSON text in a string attribute, so such a value never drops the
+    record. A field can therefore be a number in one item and a string in
+    another. An integer with more than 38 digits but at most 38 significant
+    ones (for example ``10**38``) is stored as a number.
+
     The partition key holds ``executionArn``. With the default sort key
     (``sk`` = ``emittedAt``) every export adds a new item, keeping the full
     history. With ``sort_key=None`` later exports overwrite the item.
@@ -51,9 +58,11 @@ def __init__(
                 if region
                 else boto3.client("dynamodb")
             )
-        from boto3.dynamodb.types import TypeSerializer  # deferred, same reason
+        # deferred, same reason
+        from boto3.dynamodb.types import DYNAMODB_CONTEXT, TypeSerializer
 
         self._serializer = TypeSerializer()
+        self._context = DYNAMODB_CONTEXT
 
     def render(self, record: dict[str, Any]) -> dict[str, Any]:
         return with_operations_by_name(record)
@@ -68,8 +77,35 @@ def export(self, record: dict[str, Any]) -> None:
     def flush(self) -> None:
         return None
 
+    def _number(self, text: str) -> decimal.Decimal | str:
+        # TypeSerializer checks numbers with DYNAMODB_CONTEXT, which traps any
+        # inexact, rounded, overflowing or underflowing value. Check the same
+        # way, plus DynamoDB's own magnitude range (1E-130 to 9.99...E+125),
+        # and keep a number either would reject as its JSON text.
+        try:
+            number = self._context.create_decimal(text)
+        except decimal.DecimalException:
+            # An integer with trailing zeros, such as 10**38, has more than 38
+            # digits but few significant ones; its normalized form fits.
+            digits = text.removeprefix("-")
+            significant = digits.rstrip("0")
+            if not digits.isdigit() or len(significant) > 38:
+                return text
+            sign = text[: len(text) - len(digits)]
+            exponent = len(digits) - len(significant)
+            number = decimal.Decimal(f"{sign}{significant}E{exponent}")
+            return number if number.adjusted() <= 125 else text
+        if number and not -130 <= number.adjusted() <= 125:
+            return text
+        return decimal.Decimal(text)
+
     def _marshal(self, item: dict[str, Any]) -> dict[str, Any]:
         # DynamoDB numbers must be ``Decimal``; a JSON round trip converts every
-        # float and leaves the rest of the record untouched.
-        plain = json.loads(compact_dumps(item), parse_float=Decimal)
+        # number and leaves the rest of the record untouched.
+        plain = json.loads(
+            compact_dumps(item),
+            parse_float=self._number,
+            parse_int=self._number,
+            parse_constant=str,
+        )
         return {key: self._serializer.serialize(value) for key, value in plain.items()}
diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/exporters/lambda_log_exporter.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/exporters/lambda_log_exporter.py
index 5ddb8dfc1..e66e0c5e5 100644
--- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/exporters/lambda_log_exporter.py
+++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/exporters/lambda_log_exporter.py
@@ -5,6 +5,8 @@
 
 from __future__ import annotations
 
+import datetime
+import re
 from typing import Any
 
 from aws_durable_execution_sdk_python_telemetry_exporter.exporters._common import (
@@ -15,12 +17,43 @@
 )
 
 
+# U+2028 and U+2029 (line breaks to some tools) and lone surrogates (no UTF-8
+# encoding, so printing one raises). A JSON line holds them only inside
+# strings, where a \uXXXX escape is valid and decodes to the same text.
+_ESCAPED_IN_LINE = re.compile("[\u2028\u2029\ud800-\udfff]")
+
+
+def _escape(match: re.Match[str]) -> str:
+    return f"\\u{ord(match.group()):04x}"
+
+
+def _line(rendered: Any) -> str:
+    return _ESCAPED_IN_LINE.sub(_escape, compact_dumps(rendered))
+
+
 class LambdaLogExporter:
-    """Writes ``operationsByName`` records to the function's own log group via ``print``.
+    """Writes ``operationsByName`` records to the function's own log group.
+
+    Port of the JS ``LambdaLogExporter``, which calls ``console.log(JSON.stringify(
+    withOperationsByName(record)))``. In the Lambda JSON log format the Node.js
+    runtime wraps that line in a ``{timestamp, level, requestId, message}``
+    envelope, and the record readers for this destination query
+    ``level = "INFO"`` and parse ``message``. The Python runtime captures
+    ``print`` output as plain text in both log formats, so this exporter writes
+    the envelope itself: one stdout line of compact JSON with the keys
+    ``timestamp`` (UTC ISO-8601, millisecond precision, ``Z``), ``level``
+    (``"INFO"``) and ``message`` (the compact record JSON, as a string). The
+    request id is not available to exporters, so ``requestId`` is omitted.
+    Requires no extra IAM.
 
-    Port of the JS ``LambdaLogExporter``: ``console.log(JSON.stringify(
-    withOperationsByName(record)))``. Requires no extra IAM. Emits the name-keyed
-    summary map (``OPERATIONS_BY_NAME``).
+    ``render`` returns this envelope, so ``max_record_size_bytes`` (default
+    256,000) limits the printed line itself (in UTF-8 bytes, without the
+    trailing newline). Truncation runs once, against the envelope, including
+    the JSON escaping of ``message``; ``measure`` returns the size of that
+    exact line. U+2028 and U+2029 are written as ``\\u2028`` and ``\\u2029``
+    so tools that split on them keep one record per line, and a lone surrogate
+    is written as its ``\\uXXXX`` escape so printing it never raises.
+    CloudWatch Logs accepts events of up to 1 MB.
     """
 
     def __init__(self, max_record_size_bytes: int | None = None) -> None:
@@ -29,13 +62,29 @@ def __init__(self, max_record_size_bytes: int | None = None) -> None:
         )
 
     def render(self, record: dict[str, Any]) -> dict[str, Any]:
-        return with_operations_by_name(record)
+        timestamp = (
+            datetime.datetime.now(datetime.UTC)
+            .isoformat(timespec="milliseconds")
+            .replace("+00:00", "Z")
+        )
+        return {
+            "timestamp": timestamp,
+            "level": "INFO",
+            "message": compact_dumps(with_operations_by_name(record)),
+        }
+
+    def measure(self, rendered: Any) -> int | None:
+        """Return the UTF-8 size of the line printed for ``rendered``."""
+        try:
+            return len(_line(rendered).encode("utf-8"))
+        except (TypeError, ValueError):
+            return None
 
     def export(self, record: dict[str, Any]) -> None:
-        # Raw JSON line to stdout -> the function's CloudWatch log group. The
-        # conformance CloudWatch sink json.loads each line (and unwraps the
-        # Lambda structured-log envelope when present).
-        print(compact_dumps(self.render(record)), flush=True)  # noqa: T201
+        line = _line(self.render(record))
+        # print (not logging): Lambda leaves stdout lines as written in both
+        # the Text and JSON log formats, so the envelope is never wrapped twice.
+        print(line, flush=True)  # noqa: T201
 
     def flush(self) -> None:
         return None
diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/plugin.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/plugin.py
index a3b64bf34..7313ea68c 100644
--- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/plugin.py
+++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/plugin.py
@@ -42,15 +42,20 @@
 
 from __future__ import annotations
 
+import base64
 import contextlib
 import datetime
+import decimal
 import json
 import math
+import re
 import threading
+import uuid
 from collections.abc import Iterator, Mapping
 from dataclasses import dataclass
-from typing import Any, Callable
+from typing import TYPE_CHECKING, Any, Callable
 
+from aws_durable_execution_sdk_python.concurrency.models import BatchResult
 from aws_durable_execution_sdk_python.plugin import (
     DurableInstrumentationPlugin,
     InvocationEndInfo,
@@ -60,6 +65,10 @@
     OperationInfo,
     OperationType,
 )
+from aws_durable_execution_sdk_python.serdes import (
+    EXTENDED_TYPES_SERDES,
+    SerDesContext,
+)
 
 from aws_durable_execution_sdk_python_telemetry_exporter._export_scheduler import (
     _ExportScheduler,
@@ -77,6 +86,9 @@
     DurableExecutionTelemetryConfig,
 )
 
+if TYPE_CHECKING:
+    from aws_durable_execution_sdk_python.lambda_service import ErrorObject
+
 
 # Maps the SDK invocation status onto the record status. A durable execution
 # suspends (PENDING) while waiting; from the execution's point of view it is
@@ -167,16 +179,194 @@ def _apply_result_override(
 ) -> Any:
     if raw_result is None:
         return None
-    try:
-        parsed = json.loads(raw_result)
-    except (json.JSONDecodeError, TypeError):
-        parsed = raw_result
+    parsed = _decode_checkpoint_result(raw_result)
     try:
         return transform(parsed)
     except Exception:  # noqa: BLE001 - untrusted transform must never break emission
         return None
 
 
+# -- operation results ----------------------------------------------------------
+
+
+def _is_extended_type_envelope(value: Any) -> bool:
+    # ExtendedTypeSerDes wraps every non-primitive value in exactly
+    # {"t": <type tag>, "v": <encoded value>} at the root.
+    return (
+        isinstance(value, dict)
+        and value.keys() == {"t", "v"}
+        and isinstance(value["t"], str)
+    )
+
+
+def _decode_checkpoint_result(raw_result: str) -> Any:
+    """Return an operation's checkpointed result as a plain JSON value.
+
+    The SDK's default serializer (``ExtendedTypeSerDes``) writes primitives and
+    primitive lists as plain JSON, and everything else as a ``{"t": ..., "v":
+    ...}`` envelope. An envelope is decoded with that default serializer and
+    converted to plain JSON values (see ``_to_json_compatible``). A payload that
+    is not an envelope is parsed JSON, or the raw string when it is not JSON;
+    its tree is only walked when it holds NaN, an infinite number, or a run of
+    16 or more digits (a possible integer beyond ``2**53 - 1``). An
+    envelope that fails to decode falls back to the parsed payload. Never
+    raises.
+
+    The hooks do not say which serializer wrote a result, so the envelope is
+    recognised by shape alone: a result written by ``JsonSerDes`` or a custom
+    ``SerDes`` that has this exact shape is decoded too. Only the core's known
+    tags are decoded, and the core may coerce ``v`` while decoding (for
+    example ``{"t": "b", "v": "yes"}`` becomes ``True``). An unknown tag fails
+    to decode and passes through unchanged.
+    """
+    non_finite = False
+
+    def parse_constant(name: str) -> float:
+        nonlocal non_finite
+        non_finite = True
+        return float(name)
+
+    def parse_float(text: str) -> float:
+        nonlocal non_finite
+        number = float(text)
+        if not math.isfinite(number):  # for example 1e999
+            non_finite = True
+        return number
+
+    try:
+        parsed = json.loads(
+            raw_result, parse_constant=parse_constant, parse_float=parse_float
+        )
+    except (ValueError, TypeError, RecursionError):
+        return raw_result
+    try:
+        if _is_extended_type_envelope(parsed):
+            try:
+                decoded = EXTENDED_TYPES_SERDES.deserialize(raw_result, SerDesContext())
+                return _to_json_compatible(decoded)
+            except Exception:  # noqa: BLE001 - fall back to the parsed payload
+                pass
+        # Plain JSON is already JSON-compatible, except NaN, Infinity and
+        # integers beyond 2**53 - 1. Such an integer has at least 16 digits;
+        # the C-speed scan may also match digits in a string or a float, which
+        # only costs a walk that leaves them unchanged.
+        if non_finite or _LONG_DIGIT_RUN.search(raw_result):
+            return _to_json_compatible(parsed)
+        return parsed
+    except Exception:  # noqa: BLE001 - one bad result never breaks the record
+        return raw_result
+
+
+# Largest integer that every JSON reader (including JavaScript) parses exactly.
+_MAX_SAFE_INTEGER = 2**53 - 1
+# 2**53 - 1 has 16 digits, so its leading digit's exponent is 15.
+_MAX_SAFE_INTEGER_ADJUSTED = 15
+_LONG_DIGIT_RUN = re.compile(r"\d{16,}")
+# DynamoDB's number range, the narrowest of the record destinations: a
+# magnitude from 1E-130 to 9.99...E+125. Its 38-significant-digit limit needs
+# no check: a float written exactly has at most 17 significant digits.
+_MAX_NUMBER_ADJUSTED = 125
+_MIN_NUMBER_ADJUSTED = -130
+
+
+def _to_json_compatible(value: Any) -> Any:
+    """Convert a decoded result into plain JSON values, without recursion.
+
+    ``int``: unchanged when ``abs(n) <= 2**53 - 1``, else ``str(n)``.
+    ``Decimal``: an int when integral and ``abs(n) <= 2**53 - 1``, else a float
+    when it round-trips exactly and is within DynamoDB's number range
+    (magnitude 1E-130 to 9.99...E+125), else ``str(value)``; NaN and Infinity
+    become ``None``. Negative zero becomes ``0``. ``datetime`` / ``date`` -> ``isoformat()``, ``UUID`` -> string,
+    ``bytes`` -> base64, ``tuple`` -> list, ``BatchResult`` -> ``to_dict()``.
+    Non-finite floats become ``None``.
+    """
+    root: list[Any] = [None]
+    # (value, target container, key in target)
+    pending: list[tuple[Any, Any, Any]] = [(value, root, 0)]
+    while pending:
+        item, target, key = pending.pop()
+        if isinstance(item, BatchResult):
+            item = item.to_dict()
+        if isinstance(item, (list, tuple)):
+            converted_list: list[Any] = [None] * len(item)
+            target[key] = converted_list
+            pending.extend(
+                (child, converted_list, index) for index, child in enumerate(item)
+            )
+        elif isinstance(item, dict):
+            converted_dict: dict[str, Any] = {}
+            target[key] = converted_dict
+            for name, child in item.items():
+                converted_dict[str(name)] = None
+                pending.append((child, converted_dict, str(name)))
+        else:
+            target[key] = _scalar_to_json_compatible(item)
+    return root[0]
+
+
+def _scalar_to_json_compatible(value: Any) -> Any:
+    if isinstance(value, float):
+        return value if math.isfinite(value) else None
+    if value is None or isinstance(value, (str, bool)):
+        return value
+    if isinstance(value, int):
+        # Beyond 2**53 - 1 a JavaScript reader would round it.
+        return value if abs(value) <= _MAX_SAFE_INTEGER else str(value)
+    if isinstance(value, decimal.Decimal):
+        if not value.is_finite():
+            return None
+        # adjusted() is checked first so Decimal("1E+1000000") never builds a
+        # million-digit int.
+        if (
+            value.adjusted() <= _MAX_SAFE_INTEGER_ADJUSTED
+            and value == value.to_integral_value()
+        ):
+            as_int = int(value)
+            if abs(as_int) <= _MAX_SAFE_INTEGER:
+                return as_int
+        if value and not (
+            _MIN_NUMBER_ADJUSTED <= value.adjusted() <= _MAX_NUMBER_ADJUSTED
+        ):
+            return str(value)
+        as_float = float(value)
+        written = decimal.Decimal(repr(as_float))
+        if math.isfinite(as_float) and written == value:
+            return as_float
+        return str(value)
+    if isinstance(value, (datetime.datetime, datetime.date)):
+        return value.isoformat()
+    if isinstance(value, uuid.UUID):
+        return str(value)
+    if isinstance(value, (bytes, bytearray, memoryview)):
+        return base64.b64encode(bytes(value)).decode("ascii")
+    msg = f"Unsupported decoded result type: {type(value)!r}"
+    raise TypeError(msg)
+
+
+# A longer error name or message keeps its first 8,192 code points followed
+# by "…". A Python str is a sequence of code points, so len() and slicing
+# count code points and never split a surrogate pair.
+_MAX_ERROR_TEXT_CHARS = 8192
+
+
+def _cap(text: Any) -> Any:
+    if isinstance(text, str) and len(text) > _MAX_ERROR_TEXT_CHARS:
+        return text[:_MAX_ERROR_TEXT_CHARS] + "\u2026"
+    return text
+
+
+def _error_shape(error: ErrorObject) -> dict[str, Any]:
+    """Return an error as ``{name, message}`` in that key order.
+
+    The plugin hooks carry an ``ErrorObject`` (type, message, data, stack
+    trace) and no cause, so records written by this SDK never have ``cause``.
+    The name is the SDK's ErrorType, otherwise unchanged. A name or message
+    longer than 8,192 code points is cut to its first 8,192 code points
+    followed by ``…``.
+    """
+    return {"name": _cap(error.type), "message": _cap(error.message)}
+
+
 @dataclass(frozen=True)
 class _EmissionSnapshot:
     """One admitted build's stable operation map and ordering token.
@@ -526,7 +716,7 @@ def _build_operations(
             if op.attempt is not None:
                 entry["attempt"] = op.attempt
             if shared._include_errors and op.error is not None:
-                entry["error"] = {"name": op.error.type, "message": op.error.message}
+                entry["error"] = _error_shape(op.error)
             if override is not None and override.result is not None:
                 value = _apply_result_override(override.result, op.result)
                 if value is not None:
@@ -592,7 +782,7 @@ def _emit(
         if output_value is not None:
             record["output"] = output_value
         if error is not None:
-            record["error"] = {"name": error.type, "message": error.message}
+            record["error"] = _error_shape(error)
         record["operations"] = self._build_operations(operations)
 
         # Admission and hand-off use one lock, but building does not. A newer
diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/truncation.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/truncation.py
index d826d2e0d..0e246a3ed 100644
--- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/truncation.py
+++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/truncation.py
@@ -4,6 +4,9 @@
 """Best-effort record size limiter.
 
 Direct port of the JS ``truncation.ts``. Drop order:
+  0. every error ``cause`` (``droppedCauses: true``). Records written by this
+     SDK never carry a cause, so this step only changes a record whose errors
+     were given a ``cause`` by other code;
   1. operation ``result`` fields, oldest operation first (each dropped op marked
      ``truncated: true``);
   2. whole operations, oldest first (``droppedOperations`` count);
@@ -12,9 +15,13 @@
 
 Identity/timeline fields are never dropped. The input record is never mutated.
 ``render`` maps the record to the exact value the exporter serializes, so the
-size check measures what is actually emitted. Byte size is measured with
-JS-compatible compact JSON (no whitespace, non-ASCII preserved) to match
-``JSON.stringify`` byte counts.
+size check measures what is actually emitted. By default the byte size is that
+of JS-compatible compact JSON (no whitespace, non-ASCII preserved), matching
+``JSON.stringify`` byte counts. An exporter that writes the rendered value
+differently passes its own ``measure`` (``LambdaLogExporter`` does, because it
+escapes some characters in the printed line). A value that cannot be measured
+(for example one holding a lone surrogate, which has no UTF-8 encoding) is
+returned unchanged, never raised.
 """
 
 from __future__ import annotations
@@ -24,24 +31,43 @@
 
 
 def json_byte_size(value: Any) -> int | None:
+    """Return the UTF-8 size of ``value`` as compact JSON, or ``None``.
+
+    ``None`` when the value is not JSON-serializable or has no UTF-8 encoding
+    (a lone surrogate raises ``UnicodeEncodeError``, a ``ValueError``).
+    """
     try:
-        return len(
-            json.dumps(value, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
-        )
+        text = json.dumps(value, separators=(",", ":"), ensure_ascii=False)
+        return len(text.encode("utf-8"))
     except (TypeError, ValueError):
         return None
 
 
+def _has_cause(holder: dict[str, Any]) -> bool:
+    error = holder.get("error")
+    return isinstance(error, dict) and "cause" in error
+
+
+def _drop_cause(holder: dict[str, Any]) -> None:
+    # ``holder`` is already a copy; the error is copied too, never mutated.
+    if _has_cause(holder):
+        holder["error"] = {
+            key: value for key, value in holder["error"].items() if key != "cause"
+        }
+
+
 def truncate_record(
     record: dict[str, Any],
     max_bytes: int | None,
     render: Callable[[dict[str, Any]], Any] | None = None,
+    measure: Callable[[Any], int | None] | None = None,
 ) -> dict[str, Any]:
     render = render or (lambda r: r)
+    measure = measure or json_byte_size
     if max_bytes is None or max_bytes <= 0:
         return record
 
-    initial = json_byte_size(render(record))
+    initial = measure(render(record))
     if initial is None or initial <= max_bytes:
         return record
 
@@ -53,6 +79,12 @@ def truncate_record(
         range(len(ops)), key=lambda i: (ops[i].get("startTime") or "\uffff", i)
     )
 
+    # Phase 0: drop every cause. The cause chain hangs off each error only, so
+    # removing the top-level ``cause`` removes the whole chain.
+    dropped_causes = _has_cause(record) or any(_has_cause(op) for op in ops)
+    for op in ops:
+        _drop_cause(op)
+
     any_result = False
     dropped_ops = 0
     dropped_input = False
@@ -62,6 +94,9 @@ def candidate() -> dict[str, Any]:
         out = dict(record)
         out["operations"] = [op for i, op in enumerate(ops) if kept[i]]
         out["truncated"] = True
+        if dropped_causes:
+            _drop_cause(out)
+            out["droppedCauses"] = True
         if dropped_ops > 0:
             out["droppedOperations"] = dropped_ops
         if dropped_input:
@@ -73,7 +108,7 @@ def candidate() -> dict[str, Any]:
         return out
 
     def fits() -> bool:
-        size = json_byte_size(render(candidate()))
+        size = measure(render(candidate()))
         return size is not None and size <= max_bytes
 
     # Phase 1: drop operation results oldest-first.
@@ -101,7 +136,13 @@ def fits() -> bool:
     if not fits() and record.get("output") is not None:
         dropped_output = True
 
-    if not any_result and dropped_ops == 0 and not dropped_input and not dropped_output:
+    if (
+        not dropped_causes
+        and not any_result
+        and dropped_ops == 0
+        and not dropped_input
+        and not dropped_output
+    ):
         return record
 
     return candidate()
diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/types.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/types.py
index 42a6db168..cd16a32bc 100644
--- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/types.py
+++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/types.py
@@ -56,6 +56,9 @@ class TelemetryExporter(Protocol):
     ``render`` maps the canonical record dict to the exact shape the exporter
     serializes (identity for array exporters, the ``operationsByName`` expansion
     for point-access exporters).
+    An exporter may also define ``measure(rendered) -> int | None``, the
+    byte size of what it writes for a rendered value; without it the size is
+    that of compact JSON.
     """
 
     max_record_size_bytes: int | None
diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_error_and_result_format.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_error_and_result_format.py
new file mode 100644
index 000000000..f2b7d4fee
--- /dev/null
+++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_error_and_result_format.py
@@ -0,0 +1,412 @@
+# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates.
+#
+# SPDX-License-Identifier: Apache-2.0
+"""Error shape and opted-in operation result decoding in emitted records."""
+
+from __future__ import annotations
+
+import base64
+import dataclasses
+import datetime
+import decimal
+import json
+import uuid
+from typing import Any
+
+import pytest
+from aws_durable_execution_sdk_python.concurrency.models import (
+    BatchItem,
+    BatchItemStatus,
+    BatchResult,
+    CompletionReason,
+)
+from aws_durable_execution_sdk_python.lambda_service import (
+    ErrorObject,
+    OperationStatus,
+    OperationSubType,
+)
+from aws_durable_execution_sdk_python.plugin import (
+    InvocationEndInfo,
+    InvocationStartInfo,
+    InvocationStatus,
+    OperationEndInfo,
+    OperationInfo,
+    OperationType,
+)
+from aws_durable_execution_sdk_python.serdes import EXTENDED_TYPES_SERDES
+
+from aws_durable_execution_sdk_python_telemetry_exporter import (
+    ContentConfig,
+    ContentOperations,
+    DurableExecutionTelemetryConfig,
+    OperationOverride,
+    durable_execution_telemetry,
+)
+from aws_durable_execution_sdk_python_telemetry_exporter.operations_index import (
+    with_operations_by_name,
+)
+from aws_durable_execution_sdk_python_telemetry_exporter.plugin import (
+    _decode_checkpoint_result,
+    _to_json_compatible,
+)
+
+ARN = "arn:aws:lambda:us-west-2:123456789012:function:my-fn:$LATEST/durable-execution/exec-1/inv-1"
+T0 = datetime.datetime(2026, 1, 1, 0, 0, 0, tzinfo=datetime.UTC)
+T1 = datetime.datetime(2026, 1, 1, 0, 0, 1, tzinfo=datetime.UTC)
+STEP_ERROR = "aws_durable_execution_sdk_python.exceptions.StepError"
+
+
+class CaptureExporter:
+    def __init__(self) -> None:
+        self.max_record_size_bytes: int | None = None
+        self.records: list[dict[str, Any]] = []
+
+    def render(self, record: dict[str, Any]) -> Any:  # noqa: PLR6301
+        return record
+
+    def export(self, record: dict[str, Any]) -> None:
+        self.records.append(record)
+
+    def flush(self) -> None:
+        return None
+
+
+def _op(name: str, *, result=None, error=None, status=OperationStatus.SUCCEEDED):
+    return OperationEndInfo(
+        operation_id=name,
+        operation_type=OperationType.STEP,
+        sub_type=OperationSubType.STEP,
+        name=name,
+        parent_id=None,
+        start_time=T0,
+        is_replayed=False,
+        status=status,
+        end_time=T1,
+        result=result,
+        error=error,
+        attempt=1,
+    )
+
+
+def _emit(
+    ops: list[OperationInfo],
+    *,
+    error: ErrorObject | None = None,
+    content: ContentConfig | None = None,
+) -> dict[str, Any]:
+    exporter = CaptureExporter()
+    factory = durable_execution_telemetry(
+        DurableExecutionTelemetryConfig(exporters=[exporter], content=content)
+    )
+    operations = {op.operation_id: op for op in ops}
+    start = InvocationStartInfo(
+        request_id=None,
+        execution_arn=ARN,
+        is_first_invocation=True,
+        execution_start_time=T0,
+        execution_input=None,
+        operations=operations,
+    )
+    plugin = factory.create_plugin(start)
+    plugin.on_invocation_start(start)
+    plugin.on_invocation_end(
+        InvocationEndInfo(
+            request_id=None,
+            execution_arn=ARN,
+            is_first_invocation=True,
+            execution_start_time=T0,
+            status=InvocationStatus.FAILED if error else InvocationStatus.SUCCEEDED,
+            error=error,
+            execution_result=None,
+            operations=operations,
+        )
+    )
+    assert len(exporter.records) == 1
+    return exporter.records[0]
+
+
+def _err(type_name: str | None, message: str | None, data: str | None = None):
+    return ErrorObject(message=message, type=type_name, data=data, stack_trace=None)
+
+
+# -- error shape ---------------------------------------------------------------
+
+
+def test_hook_errors_carry_no_cause():
+    # The plugin hooks expose ErrorObject only; it has no cause field, so a
+    # Python record can never carry ``cause``.
+    fields = {f.name for f in dataclasses.fields(ErrorObject)}
+    assert fields == {"message", "type", "data", "stack_trace"}
+    assert "error" in {f.name for f in dataclasses.fields(InvocationEndInfo)}
+    assert "error" in {f.name for f in dataclasses.fields(OperationEndInfo)}
+
+
+def test_error_shape_and_key_order_everywhere():
+    record = _emit(
+        [
+            _op(
+                "charge",
+                status=OperationStatus.FAILED,
+                error=_err("PaymentError", "card declined", data='{"code":1}'),
+            )
+        ],
+        error=_err(STEP_ERROR, "card declined", data="ignored"),
+    )
+    assert list(record["error"]) == ["name", "message"]
+    # Names are the SDK's ErrorType, unchanged.
+    assert record["error"] == {"name": STEP_ERROR, "message": "card declined"}
+    op_error = record["operations"][0]["error"]
+    assert list(op_error) == ["name", "message"]
+    assert op_error == {"name": "PaymentError", "message": "card declined"}
+    by_name = with_operations_by_name(record)["operationsByName"]["charge"]["error"]
+    assert list(by_name) == ["name", "message"]
+    assert by_name == op_error
+    # ErrorData is never turned into a cause.
+    assert "cause" not in json.dumps(record)
+
+
+def test_error_with_missing_type_and_message_keeps_both_keys():
+    record = _emit([], error=_err(None, None))
+    assert list(record["error"]) == ["name", "message"]
+    assert record["error"] == {"name": None, "message": None}
+
+
+def test_include_errors_false_removes_operation_errors_everywhere():
+    record = _emit(
+        [_op("charge", status=OperationStatus.FAILED, error=_err("E", "m"))],
+        error=_err(STEP_ERROR, "m"),
+        content=ContentConfig(operations=ContentOperations(include_errors=False)),
+    )
+    assert "error" not in record["operations"][0]
+    assert "error" not in with_operations_by_name(record)["operationsByName"]["charge"]
+    assert record["error"] == {"name": STEP_ERROR, "message": "m"}
+
+
+# -- result decoding -----------------------------------------------------------
+
+
+def _results(*ops: OperationInfo) -> dict[str, Any]:
+    overrides = [OperationOverride(op.name or "", result=lambda r: r) for op in ops]
+    record = _emit(
+        list(ops),
+        content=ContentConfig(operations=ContentOperations(overrides=overrides)),
+    )
+    return {op["name"]: op.get("result") for op in record["operations"]}
+
+
+def _envelope(value: Any) -> str:
+    raw = EXTENDED_TYPES_SERDES.serialize(value, None)  # type: ignore[arg-type]
+    assert set(json.loads(raw)) == {"t", "v"}
+    return raw
+
+
+def test_envelope_result_decodes_to_plain_json():
+    uid = uuid.UUID("12345678-1234-5678-1234-567812345678")
+    value = {
+        "amount": decimal.Decimal("12.50"),
+        "count": decimal.Decimal(3),
+        "when": datetime.datetime(2026, 1, 2, 3, 4, 5, 6000, tzinfo=datetime.UTC),
+        "day": datetime.date(2026, 1, 2),
+        "id": uid,
+        "blob": b"\x00\xffhi",
+        "pair": (1, "a"),
+        "nested": [{"x": decimal.Decimal("0.1")}],
+    }
+    results = _results(_op("r", result=_envelope(value)))
+    assert results["r"] == {
+        "amount": 12.5,
+        "count": 3,
+        "when": "2026-01-02T03:04:05.006000+00:00",
+        "day": "2026-01-02",
+        "id": str(uid),
+        "blob": base64.b64encode(b"\x00\xffhi").decode(),
+        "pair": [1, "a"],
+        "nested": [{"x": 0.1}],
+    }
+    json.dumps(results["r"], allow_nan=False)
+
+
+def test_batch_result_becomes_to_dict():
+    batch = BatchResult(
+        all=[BatchItem(0, BatchItemStatus.SUCCEEDED, result=decimal.Decimal(2))],
+        completion_reason=CompletionReason.ALL_COMPLETED,
+    )
+    decoded = _decode_checkpoint_result(_envelope(batch))
+    expected = batch.to_dict()
+    expected["all"][0]["result"] = 2
+    assert decoded == json.loads(json.dumps(expected))
+
+
+@pytest.mark.parametrize(
+    ("text", "expected"),
+    [
+        ("12", 12),
+        ("-0", 0),
+        ("-0.0", 0),
+        ("1.10", 1.1),
+        ("0.1", 0.1),
+        ("9007199254740991", 9007199254740991),
+        ("-9007199254740991", -9007199254740991),
+        ("9007199254740992", 9007199254740992.0),
+        ("9007199254740993", "9007199254740993"),
+        ("1E+20", 1e20),
+        ("1E+125", 1e125),
+        ("1E+126", "1E+126"),
+        ("1E-130", 1e-130),
+        ("1E-131", "1E-131"),
+        ("1E+299", "1E+299"),
+        ("1E+300", "1E+300"),
+        ("1" * 39, "1" * 39),
+        ("12345678901234567.89", "12345678901234567.89"),
+        ("1E+5000", "1E+5000"),
+        ("1E+1000000", "1E+1000000"),
+        ("1E-1000000", "1E-1000000"),
+        ("NaN", None),
+        ("sNaN", None),
+        ("Infinity", None),
+        ("-Infinity", None),
+    ],
+)
+def test_decimal_conversion(text, expected):
+    raw = _envelope({"v": decimal.Decimal(text)})
+    decoded = _decode_checkpoint_result(raw)
+    assert decoded == {"v": expected}
+    assert type(decoded["v"]) is type(expected)
+
+
+class _CountingDecimal(decimal.Decimal):
+    int_calls = 0
+
+    def __int__(self) -> int:
+        type(self).int_calls += 1
+        return super().__int__()
+
+
+@pytest.mark.parametrize(
+    "text", ["1E+16", "1E+299", "1E+5000", "1E+1000000", "1" * 4000]
+)
+def test_out_of_range_integral_decimal_never_builds_an_int(text):
+    # Counts int() conversions instead of timing them: a Decimal past the
+    # 2**53 - 1 bound is rejected on its exponent before any int is built.
+    _CountingDecimal.int_calls = 0
+    assert _to_json_compatible([_CountingDecimal(text)]) == [
+        float(text) if text == "1E+16" else text
+    ]
+    assert _CountingDecimal.int_calls == 0
+    assert _to_json_compatible([_CountingDecimal("12")]) == [12]
+    assert _CountingDecimal.int_calls == 1
+
+
+def test_non_finite_plain_json_becomes_null():
+    assert _decode_checkpoint_result("[NaN, Infinity, 1.5]") == [None, None, 1.5]
+    assert _decode_checkpoint_result('{"a": [1e999, -1e999]}') == {"a": [None, None]}
+
+
+def test_plain_json_skips_the_tree_walk(monkeypatch):
+    from aws_durable_execution_sdk_python_telemetry_exporter import plugin
+
+    calls: list[Any] = []
+    real = plugin._to_json_compatible
+
+    def spy(value):
+        calls.append(value)
+        return real(value)
+
+    monkeypatch.setattr(plugin, "_to_json_compatible", spy)
+    assert _decode_checkpoint_result('{"a": [1, 2.5, "x", null, true]}') == {
+        "a": [1, 2.5, "x", None, True]
+    }
+    assert _decode_checkpoint_result('{"t": "zz", "v": 1}') == {"t": "zz", "v": 1}
+    assert calls == []
+    assert _decode_checkpoint_result("[NaN]") == [None]
+    assert len(calls) == 1
+
+
+def test_600_nested_arrays():
+    depth = 600
+    raw = "[" * depth + "]" * depth
+    decoded = _decode_checkpoint_result(raw)
+    level = 1
+    while decoded:
+        decoded = decoded[0]
+        level += 1
+    assert level == depth
+
+    # Converted without recursion, also for tuples.
+    value: Any = ()
+    for _ in range(depth - 1):
+        value = (value, decimal.Decimal("1.5"))
+    converted = _to_json_compatible(value)
+    level = 1
+    while converted:
+        assert converted[1] == 1.5
+        converted = converted[0]
+        level += 1
+    assert level == depth
+
+
+def test_deep_envelope_the_core_cannot_decode_does_not_raise():
+    depth = 600
+    raw = '{"t":"l","v":[' * depth + "]}" * depth
+    decoded = _decode_checkpoint_result(raw)
+    # json.loads or the core decoder gives up; the raw/parsed value is kept.
+    assert decoded is not None
+
+
+def test_non_envelope_results_pass_through():
+    assert _decode_checkpoint_result("42") == 42
+    assert _decode_checkpoint_result('{"a": 1}') == {"a": 1}
+    assert _decode_checkpoint_result("not json") == "not json"
+    # Unknown tag: decoding fails, parsed value kept.
+    assert _decode_checkpoint_result('{"t":"zz","v":1}') == {"t": "zz", "v": 1}
+    # Extra key: not an envelope.
+    assert _decode_checkpoint_result('{"t":"s","v":"x","w":1}') == {
+        "t": "s",
+        "v": "x",
+        "w": 1,
+    }
+    # Shape collision: a JsonSerDes value that looks like an envelope is decoded.
+    assert _decode_checkpoint_result('{"t":"s","v":"x"}') == "x"
+    assert _decode_checkpoint_result('{"t":"i","v":5}') == 5
+    assert _decode_checkpoint_result('{"t":"m","v":{"a":1}}') == {"a": 1}
+    # The core coerces ``v`` for a known tag.
+    assert _decode_checkpoint_result('{"t":"b","v":"yes"}') is True
+    assert _decode_checkpoint_result('{"t":"temperature","v":21.5}') == {
+        "t": "temperature",
+        "v": 21.5,
+    }
+
+
+def test_one_failing_result_keeps_the_rest_of_the_record():
+    results = _results(
+        _op("good", result=_envelope({"a": decimal.Decimal("1.5")})),
+        _op("bad", result='{"t":"d","v":"not-a-decimal"}'),
+        _op("text", result="plain text"),
+    )
+    assert results == {
+        "good": {"a": 1.5},
+        "bad": {"t": "d", "v": "not-a-decimal"},
+        "text": "plain text",
+    }
+
+
+def test_unsupported_decoded_type_falls_back_to_raw(monkeypatch):
+    class Odd:
+        pass
+
+    monkeypatch.setattr(
+        EXTENDED_TYPES_SERDES, "deserialize", lambda data, ctx=None: {"x": Odd()}
+    )
+    raw = '{"t":"m","v":{}}'
+    # Decoded value cannot be converted; the parsed payload is kept.
+    assert _decode_checkpoint_result(raw) == {"t": "m", "v": {}}
+
+
+def test_unsupported_value_after_fallback_returns_raw_string(monkeypatch):
+    from aws_durable_execution_sdk_python_telemetry_exporter import plugin
+
+    def boom(value):
+        raise TypeError
+
+    monkeypatch.setattr(plugin, "_to_json_compatible", boom)
+    assert _decode_checkpoint_result('{"a":NaN}') == '{"a":NaN}'
diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_exporters.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_exporters.py
index 87024608f..f8373e12a 100644
--- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_exporters.py
+++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_exporters.py
@@ -10,7 +10,9 @@
 
 from __future__ import annotations
 
+import datetime
 import json
+import re
 from typing import Any
 
 import pytest
@@ -32,6 +34,9 @@
 from aws_durable_execution_sdk_python_telemetry_exporter.exporters.s3_exporter import (
     S3Partitioning as S3PartitioningFromModule,
 )
+from aws_durable_execution_sdk_python_telemetry_exporter.operations_index import (
+    with_operations_by_name,
+)
 
 
 def _record(**kw: Any) -> dict[str, Any]:
@@ -88,22 +93,45 @@ def test_every_exporter_is_re_exported_from_the_package_root():
 # -- LambdaLogExporter --------------------------------------------------------
 
 
-def test_lambda_log_default_size_and_render_is_operations_by_name():
+def test_lambda_log_default_size_and_render_is_the_envelope():
     exporter = LambdaLogExporter()
     assert exporter.max_record_size_bytes == 256_000
-    shaped = exporter.render(_record())
+    envelope = exporter.render(_record())
+    # render returns the printed envelope, so truncation measures the line.
+    assert list(envelope) == ["timestamp", "level", "message"]
+    shaped = json.loads(envelope["message"])
     assert "operations" not in shaped
     assert shaped["operationsByName"]["greet"]["count"] == 1
 
 
-def test_lambda_log_export_prints_compact_json_line(capsys):
-    LambdaLogExporter().export(_record())
-    out = capsys.readouterr().out.strip()
+def test_lambda_log_export_prints_one_envelope_line(capsys):
+    exporter = LambdaLogExporter()
+    before = datetime.datetime.now(datetime.UTC)
+    exporter.export(_record())
+    after = datetime.datetime.now(datetime.UTC)
+    out = capsys.readouterr().out
+    assert out.endswith("\n") and out.count("\n") == 1
+    line = out.rstrip("\n")
     # single compact JSON line, no whitespace separators
-    assert ", " not in out and '": ' not in out
-    parsed = json.loads(out)
-    assert parsed["recordType"] == "DurableExecutionTelemetry"
-    assert "operationsByName" in parsed
+    assert ", " not in line and '": ' not in line
+    envelope = json.loads(line)
+    assert list(envelope) == ["timestamp", "level", "message"]
+    assert envelope["level"] == "INFO"
+    # message parses back to exactly the rendered record
+    assert isinstance(envelope["message"], str)
+    expected = with_operations_by_name(_record())
+    assert json.loads(envelope["message"]) == expected
+    assert envelope["message"] == json.dumps(
+        expected, separators=(",", ":"), ensure_ascii=False
+    )
+    assert envelope["message"] == exporter.render(_record())["message"]
+    # UTC, millisecond precision, Z suffix
+    ts = envelope["timestamp"]
+    assert re.fullmatch(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z", ts)
+    parsed_ts = datetime.datetime.fromisoformat(ts.replace("Z", "+00:00"))
+    assert parsed_ts.utcoffset() == datetime.timedelta(0)
+    assert before.replace(microsecond=before.microsecond // 1000 * 1000) <= parsed_ts
+    assert parsed_ts <= after
 
 
 def test_lambda_log_custom_size_and_flush_noop():
diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_plugin.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_plugin.py
index 673a4d1c2..0532fc39f 100644
--- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_plugin.py
+++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_plugin.py
@@ -676,10 +676,12 @@ def test_default_exporter_actually_emits_to_stdout(capsys):
     factory = durable_execution_telemetry(DurableExecutionTelemetryConfig())
     _run(factory, ops=[_step("greet")])
     out = capsys.readouterr().out
-    assert (
-        '"recordType":"DurableExecutionTelemetry"' in out
-    )  # compact JSON via LambdaLogExporter
-    assert '"operationsByName"' in out
+    # compact JSON envelope via LambdaLogExporter; the record is its message
+    envelope = json.loads(out.strip())
+    assert envelope["level"] == "INFO"
+    record = json.loads(envelope["message"])
+    assert record["recordType"] == "DurableExecutionTelemetry"
+    assert "operationsByName" in record
 
 
 # -- terminal timestamps: end time only for SUCCEEDED/FAILED (comment 3) ------
diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_public_path_limits.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_public_path_limits.py
new file mode 100644
index 000000000..063eae1a3
--- /dev/null
+++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_public_path_limits.py
@@ -0,0 +1,468 @@
+# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates.
+#
+# SPDX-License-Identifier: Apache-2.0
+"""Public plugin path into real exporters: decoded numbers, line size, caps."""
+
+from __future__ import annotations
+
+import datetime
+import decimal
+import io
+import json
+import logging
+import sys
+from typing import Any
+
+import pytest
+from aws_durable_execution_sdk_python.lambda_service import (
+    ErrorObject,
+    OperationStatus,
+    OperationSubType,
+)
+from aws_durable_execution_sdk_python.plugin import (
+    InvocationEndInfo,
+    InvocationStartInfo,
+    InvocationStatus,
+    OperationEndInfo,
+    OperationInfo,
+    OperationType,
+)
+from aws_durable_execution_sdk_python.serdes import EXTENDED_TYPES_SERDES
+
+from aws_durable_execution_sdk_python_telemetry_exporter import (
+    ContentConfig,
+    ContentOperations,
+    DurableExecutionTelemetryConfig,
+    DynamoDBExporter,
+    LambdaLogExporter,
+    OperationOverride,
+    durable_execution_telemetry,
+)
+
+ARN = "arn:aws:lambda:us-west-2:123456789012:function:my-fn:$LATEST/durable-execution/exec-1/inv-1"
+T0 = datetime.datetime(2026, 1, 1, 0, 0, 0, tzinfo=datetime.UTC)
+T1 = datetime.datetime(2026, 1, 1, 0, 0, 1, tzinfo=datetime.UTC)
+BIG_39_DIGITS = "123456789012345678901234567890123456789"
+
+
+class StubDynamoDBClient:
+    def __init__(self) -> None:
+        self.items: list[dict[str, Any]] = []
+
+    def put_item(self, **kwargs: Any) -> None:
+        self.items.append(kwargs["Item"])
+
+
+def _op(name: str, index: int, *, result: str | None = None) -> OperationEndInfo:
+    start = T0 + datetime.timedelta(milliseconds=index)
+    return OperationEndInfo(
+        operation_id=f"{name}-id",
+        operation_type=OperationType.STEP,
+        sub_type=OperationSubType.STEP,
+        name=name,
+        parent_id=None,
+        start_time=start,
+        is_replayed=False,
+        status=OperationStatus.SUCCEEDED,
+        end_time=T1,
+        result=result,
+        error=None,
+        attempt=1,
+    )
+
+
+def _run(
+    exporter: Any,
+    ops: list[OperationEndInfo],
+    *,
+    execution_input: str | None = None,
+    error: ErrorObject | None = None,
+) -> None:
+    overrides = [OperationOverride(op.name or "", result=lambda r: r) for op in ops]
+    factory = durable_execution_telemetry(
+        DurableExecutionTelemetryConfig(
+            exporters=[exporter],
+            content=ContentConfig(operations=ContentOperations(overrides=overrides)),
+        )
+    )
+    operations: dict[str, OperationInfo] = {op.operation_id: op for op in ops}
+    start = InvocationStartInfo(
+        request_id=None,
+        execution_arn=ARN,
+        is_first_invocation=True,
+        execution_start_time=T0,
+        execution_input=execution_input,
+        operations=operations,
+    )
+    plugin = factory.create_plugin(start)
+    plugin.on_invocation_start(start)
+    plugin.on_invocation_end(
+        InvocationEndInfo(
+            request_id=None,
+            execution_arn=ARN,
+            is_first_invocation=True,
+            execution_start_time=T0,
+            status=InvocationStatus.FAILED if error else InvocationStatus.SUCCEEDED,
+            error=error,
+            execution_result=None,
+            operations=operations,
+        )
+    )
+
+
+def _envelope(value: Any) -> str:
+    return EXTENDED_TYPES_SERDES.serialize(value, None)  # type: ignore[arg-type]
+
+
+# -- B1: decoded numbers reach DynamoDB ----------------------------------------
+
+
+def _dynamodb_results(*results: str, caplog: pytest.LogCaptureFixture) -> Any:
+    client = StubDynamoDBClient()
+    exporter = DynamoDBExporter(table_name="t", client=client)
+    ops = [_op(f"op{i}", i, result=raw) for i, raw in enumerate(results)]
+    with caplog.at_level(logging.WARNING):
+        _run(exporter, ops)
+    assert "failed" not in caplog.text
+    assert len(client.items) == 1
+    by_name = client.items[0]["operationsByName"]["M"]
+    return [by_name[f"op{i}"]["M"]["result"] for i in range(len(results))]
+
+
+def test_out_of_range_decoded_decimals_keep_the_dynamodb_item(caplog):
+    results = _dynamodb_results(
+        _envelope(decimal.Decimal("1E+299")),
+        _envelope(decimal.Decimal(BIG_39_DIGITS)),
+        _envelope(decimal.Decimal("9007199254740993")),
+        _envelope({"amount": decimal.Decimal("12.50"), "n": decimal.Decimal("7")}),
+        caplog=caplog,
+    )
+    assert results[:3] == [
+        {"S": "1E+299"},
+        {"S": BIG_39_DIGITS},
+        # Exact for JavaScript readers, which would round the number.
+        {"S": "9007199254740993"},
+    ]
+    assert results[3] == {"M": {"amount": {"N": "12.5"}, "n": {"N": "7"}}}
+
+
+def test_out_of_range_plain_json_numbers_keep_the_dynamodb_item(caplog):
+    results = _dynamodb_results(
+        "1e300",
+        BIG_39_DIGITS,
+        "[1e-200, 1.5, 3]",
+        caplog=caplog,
+    )
+    assert results == [
+        {"S": "1e+300"},
+        {"S": BIG_39_DIGITS},
+        {"L": [{"S": "1e-200"}, {"N": "1.5"}, {"N": "3"}]},
+    ]
+
+
+def test_dynamodb_marshal_turns_non_finite_numbers_into_strings():
+    client = StubDynamoDBClient()
+    exporter = DynamoDBExporter(table_name="t", client=client)
+    marshalled = exporter._marshal({"a": float("nan"), "b": float("inf"), "c": 1})
+    assert marshalled == {"a": {"S": "NaN"}, "b": {"S": "Infinity"}, "c": {"N": "1"}}
+
+
+def test_dynamodb_marshal_applies_the_service_magnitude_range():
+    exporter = DynamoDBExporter(table_name="t", client=StubDynamoDBClient())
+    marshalled = exporter._marshal({"a": [1e125, 1e126, 1e-130, 1e-131, 0, -0.0]})
+    assert marshalled["a"]["L"] == [
+        {"N": "1E+125"},
+        {"S": "1e+126"},
+        {"N": "1E-130"},
+        {"S": "1e-131"},
+        {"N": "0"},
+        {"N": "-0.0"},
+    ]
+
+
+# -- B2: LambdaLogExporter bounds the printed line ------------------------------
+
+
+def _printed_line(capsys: pytest.CaptureFixture[str]) -> str:
+    out = capsys.readouterr().out
+    assert out.endswith("\n") and out.count("\n") == 1
+    return out[:-1]
+
+
+def test_quote_heavy_record_line_fits_the_limit(capsys):
+    limit = 262_144
+    exporter = LambdaLogExporter(max_record_size_bytes=limit)
+    # Each '"' is 2 bytes in the record JSON and 4 bytes in the printed line.
+    ops = [_op(f"q{i}", i, result=json.dumps('"' * 4000)) for i in range(40)]
+    _run(exporter, ops, execution_input='"\\' * 10_000)
+    line = _printed_line(capsys)
+    assert len(line.encode("utf-8")) <= limit
+    envelope = json.loads(line)
+    record = json.loads(envelope["message"])
+    assert record["truncated"] is True
+    # The single pass measured the line, so it still keeps what fits.
+    kept = [summary for summary in record["operationsByName"].values()]
+    assert kept
+    assert any("result" in summary for summary in kept)
+
+
+def test_capped_execution_error_fits_a_64_kb_line(capsys):
+    limit = 65_536
+    exporter = LambdaLogExporter(max_record_size_bytes=limit)
+    # \x01 is the worst case: 6 bytes in the record, 7 in the printed line.
+    message = "\x01" * 100_000
+    ops = [_op(f"q{i}", i, result=json.dumps('"' * 2000)) for i in range(40)]
+    _run(
+        exporter,
+        ops,
+        execution_input='"' * 50_000,
+        error=ErrorObject(message=message, type="E", data=None, stack_trace=None),
+    )
+    line = _printed_line(capsys)
+    assert len(line.encode("utf-8")) <= limit
+    record = json.loads(json.loads(line)["message"])
+    assert record["error"]["message"] == "\x01" * 8192 + "\u2026"
+    assert record["droppedInput"] is True
+
+
+def test_line_separators_are_escaped_and_text_round_trips(capsys):
+    exporter = LambdaLogExporter()
+    text = 'a\u2028b\u2029c é "q" \\ \n \x01 \U0001f600'
+    _run(exporter, [_op("s", 0, result=json.dumps(text))])
+    out = capsys.readouterr().out
+    assert "\u2028" not in out and "\u2029" not in out
+    assert len(out.splitlines()) == 1
+    # Non-ASCII stays raw; control characters and quotes are escaped.
+    assert "é" in out and "\U0001f600" in out and "\x01" not in out
+    record = json.loads(json.loads(out)["message"])
+    assert record["operationsByName"]["s"]["result"] == text
+
+
+def test_line_separator_escaping_is_counted_in_the_limit(capsys):
+    limit = 4096
+    exporter = LambdaLogExporter(max_record_size_bytes=limit)
+    ops = [_op(f"s{i}", i, result=json.dumps("\u2028" * 300)) for i in range(10)]
+    _run(exporter, ops)
+    line = _printed_line(capsys)
+    assert len(line.encode("utf-8")) <= limit
+    assert json.loads(json.loads(line)["message"])["truncated"] is True
+
+
+# -- rule 7: message cap --------------------------------------------------------
+
+
+def test_error_messages_are_capped_at_8192_characters(capsys):
+    exporter = LambdaLogExporter(max_record_size_bytes=0)
+    op = OperationEndInfo(
+        operation_id="f-id",
+        operation_type=OperationType.STEP,
+        sub_type=OperationSubType.STEP,
+        name="f",
+        parent_id=None,
+        start_time=T0,
+        is_replayed=False,
+        status=OperationStatus.FAILED,
+        end_time=T1,
+        result=None,
+        error=ErrorObject(message="m" * 8193, type="E", data=None, stack_trace=None),
+        attempt=1,
+    )
+    _run(
+        exporter,
+        [op],
+        error=ErrorObject(message="x" * 8192, type="E", data=None, stack_trace=None),
+    )
+    record = json.loads(json.loads(_printed_line(capsys))["message"])
+    # Exactly 8,192 characters is kept as is.
+    assert record["error"]["message"] == "x" * 8192
+    assert record["operationsByName"]["f"]["error"]["message"] == "m" * 8192 + "…"
+
+
+def test_error_names_are_capped_at_8192_code_points(capsys):
+    exporter = LambdaLogExporter(max_record_size_bytes=0)
+    _run(
+        exporter,
+        [],
+        error=ErrorObject(
+            message="\U0001f600" * 8193, type="N" * 70_000, data=None, stack_trace=None
+        ),
+    )
+    record = json.loads(json.loads(_printed_line(capsys))["message"])
+    assert record["error"]["name"] == "N" * 8192 + "\u2026"
+    # Code points, not UTF-16 units: 8,192 whole emoji are kept.
+    assert record["error"]["message"] == "\U0001f600" * 8192 + "\u2026"
+
+
+def test_long_error_name_fits_a_64_kb_line(capsys):
+    limit = 65_536
+    exporter = LambdaLogExporter(max_record_size_bytes=limit)
+    ops = [_op(f"q{i}", i, result=json.dumps('"' * 2000)) for i in range(20)]
+    _run(
+        exporter,
+        ops,
+        error=ErrorObject(
+            message="m" * 100, type="N" * 70_000, data=None, stack_trace=None
+        ),
+    )
+    line = _printed_line(capsys)
+    assert len(line.encode("utf-8")) <= limit
+    record = json.loads(json.loads(line)["message"])
+    assert record["error"]["name"] == "N" * 8192 + "\u2026"
+
+
+def test_worst_case_capped_name_and_message_fit_a_128_kb_line(capsys):
+    limit = 131_072
+    exporter = LambdaLogExporter(max_record_size_bytes=limit)
+    # \x01 is the worst case for both: 7 bytes each in the printed line.
+    ops = [_op(f"q{i}", i, result=json.dumps('"' * 2000)) for i in range(40)]
+    _run(
+        exporter,
+        ops,
+        execution_input='"' * 50_000,
+        error=ErrorObject(
+            message="\x01" * 70_000, type="\x01" * 70_000, data=None, stack_trace=None
+        ),
+    )
+    line = _printed_line(capsys)
+    assert len(line.encode("utf-8")) <= limit
+    record = json.loads(json.loads(line)["message"])
+    assert record["error"] == {
+        "name": "\x01" * 8192 + "\u2026",
+        "message": "\x01" * 8192 + "\u2026",
+    }
+    assert record["droppedInput"] is True
+
+
+# -- N-B1: a lone surrogate never drops the record -----------------------------
+
+
+def test_lone_surrogate_in_error_message_keeps_the_dynamodb_item(caplog):
+    client = StubDynamoDBClient()
+    exporter = DynamoDBExporter(table_name="t", client=client)
+    with caplog.at_level(logging.WARNING):
+        _run(
+            exporter,
+            [],
+            error=ErrorObject(
+                message="a\ud800b", type="E", data=None, stack_trace=None
+            ),
+        )
+    assert "failed" not in caplog.text
+    assert len(client.items) == 1
+    assert client.items[0]["error"]["M"]["message"] == {"S": "a\ud800b"}
+
+
+def test_lone_surrogate_is_escaped_in_the_printed_line(monkeypatch, caplog):
+    # A strict UTF-8 stdout, as in a real process; printing a raw lone
+    # surrogate to it raises UnicodeEncodeError.
+    raw = io.BytesIO()
+    stdout = io.TextIOWrapper(raw, encoding="utf-8", errors="strict")
+    monkeypatch.setattr(sys, "stdout", stdout)
+    with caplog.at_level(logging.WARNING):
+        _run(
+            LambdaLogExporter(),
+            [],
+            error=ErrorObject(
+                message="a\ud800b", type="E", data=None, stack_trace=None
+            ),
+        )
+    stdout.flush()
+    assert "failed" not in caplog.text
+    line = raw.getvalue().decode("utf-8")
+    assert line.count("\n") == 1
+    assert "a\\ud800b" in line
+    record = json.loads(json.loads(line)["message"])
+    assert record["error"]["message"] == "a\ud800b"
+
+
+def test_lone_surrogates_are_counted_in_the_line_limit(capsys):
+    limit = 4096
+    exporter = LambdaLogExporter(max_record_size_bytes=limit)
+    ops = [_op(f"s{i}", i, result=json.dumps("\ud800" * 300)) for i in range(10)]
+    _run(exporter, ops)
+    line = _printed_line(capsys)
+    assert len(line.encode("utf-8")) <= limit
+    assert json.loads(json.loads(line)["message"])["truncated"] is True
+
+
+# -- non-blocking 1: sizes are exact per exporter -------------------------------
+
+
+def test_line_separators_are_counted_raw_for_other_exporters(caplog):
+    client = StubDynamoDBClient()
+    # 60,000 U+2028 are 180,000 bytes raw and 360,000 bytes escaped.
+    exporter = DynamoDBExporter(
+        table_name="t", client=client, max_record_size_bytes=200_000
+    )
+    with caplog.at_level(logging.WARNING):
+        _run(exporter, [_op("s", 0, result=json.dumps("\u2028" * 60_000))])
+    assert len(client.items) == 1
+    item = client.items[0]
+    assert "truncated" not in item
+    assert item["operationsByName"]["M"]["s"]["M"]["result"] == {"S": "\u2028" * 60_000}
+
+
+def test_lambda_log_measure_is_the_printed_line_size(capsys):
+    exporter = LambdaLogExporter()
+    record: dict[str, Any] = {
+        "executionArn": ARN,
+        "operations": [],
+        "error": {"name": "E", "message": '\u2028\u2029\ud800é\U0001f600"\x01'},
+    }
+    size = exporter.measure(exporter.render(record))
+    exporter.export(record)
+    assert size == len(_printed_line(capsys).encode("utf-8"))
+    assert exporter.measure({"a": object()}) is None
+
+
+# -- non-blocking 2: integers beyond 2**53 - 1 ----------------------------------
+
+
+def _results(capsys: pytest.CaptureFixture[str], *results: str) -> list[Any]:
+    ops = [_op(f"op{i}", i, result=raw) for i, raw in enumerate(results)]
+    _run(LambdaLogExporter(), ops)
+    by_name = json.loads(json.loads(_printed_line(capsys))["message"])[
+        "operationsByName"
+    ]
+    return [by_name[f"op{i}"]["result"] for i in range(len(results))]
+
+
+def test_large_ints_become_strings(capsys):
+    big = 12345678901234567
+    safe = 2**53 - 1
+    results = _results(
+        capsys,
+        _envelope({"x": big, "y": safe, "z": -big}),
+        _envelope((1, -safe)),
+        str(big),
+        json.dumps([safe, -(2**60), {"k": 2**53}]),
+        # Long digit runs in strings and floats are left alone.
+        json.dumps({"s": "12345678901234567", "f": 0.12345678901234568}),
+        str(safe),
+    )
+    assert results == [
+        {"x": "12345678901234567", "y": safe, "z": "-12345678901234567"},
+        [1, -safe],
+        "12345678901234567",
+        [safe, str(-(2**60)), {"k": str(2**53)}],
+        {"s": "12345678901234567", "f": 0.12345678901234568},
+        safe,
+    ]
+
+
+# -- non-blocking 7: integers with trailing zeros stay DynamoDB numbers ---------
+
+
+def test_dynamodb_marshal_keeps_integers_with_few_significant_digits():
+    exporter = DynamoDBExporter(table_name="t", client=StubDynamoDBClient())
+    marshalled = exporter._marshal(
+        {"a": [10**38, -(10**38), 10**125, 10**126, 10**38 + 1, 1.0e38]}
+    )
+    assert marshalled["a"]["L"] == [
+        {"N": "1E+38"},
+        {"N": "-1E+38"},
+        {"N": "1E+125"},
+        {"S": str(10**126)},
+        {"S": str(10**38 + 1)},
+        {"N": "1E+38"},
+    ]
diff --git a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_shaping.py b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_shaping.py
index 8f2c8f274..77f678aa8 100644
--- a/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_shaping.py
+++ b/packages/aws-durable-execution-sdk-python-telemetry-exporter/tests/test_shaping.py
@@ -5,6 +5,8 @@
 
 from __future__ import annotations
 
+from typing import Any
+
 from aws_durable_execution_sdk_python_telemetry_exporter.operations_index import (
     OperationsFormat,
     apply_operations_format,
@@ -160,3 +162,65 @@ def test_apply_operations_format_both_keeps_array_and_adds_map():
 
 def test_operations_format_enum_values():
     assert {f.value for f in OperationsFormat} == {"array", "by-name", "both"}
+
+
+def test_truncation_drops_every_cause_first():
+    cause = {
+        "name": "Inner",
+        "message": "c" * 400,
+        "cause": {"name": "E", "message": "x"},
+    }
+    op_error = {"name": "E", "message": "m", "cause": cause}
+    record = {
+        "executionArn": "arn",
+        "error": {"name": "E", "message": "m", "cause": cause},
+        "operations": [_op("a", error=op_error, result="r")],
+    }
+    out = truncate_record(record, 300, render=lambda r: r)
+    assert out["droppedCauses"] is True
+    assert out["error"] == {"name": "E", "message": "m"}
+    assert out["operations"][0]["error"] == {"name": "E", "message": "m"}
+    # Dropping causes was enough: later phases did not run.
+    assert out["operations"][0]["result"] == "r"
+    # The input record is never mutated.
+    assert record["error"]["cause"] is cause
+    assert record["operations"][0]["error"]["cause"] is cause
+
+
+def test_truncation_without_causes_sets_no_dropped_causes():
+    record = _record_with_results([2000])
+    out = truncate_record(record, 400, render=lambda r: r)
+    assert out["truncated"] is True
+    assert "droppedCauses" not in out
+
+
+def test_json_byte_size_is_exact_for_line_separators():
+    from aws_durable_execution_sdk_python_telemetry_exporter.truncation import (
+        json_byte_size,
+    )
+
+    # Written raw by every exporter except LambdaLogExporter: 3 bytes each.
+    assert json_byte_size({"a": "\u2028" * 10}) == 38
+    assert json_byte_size("\u2028\u2029") == len('"\u2028\u2029"'.encode())
+
+
+def test_json_byte_size_returns_none_for_a_lone_surrogate():
+    from aws_durable_execution_sdk_python_telemetry_exporter.truncation import (
+        json_byte_size,
+    )
+
+    assert json_byte_size({"a": "a\ud800b"}) is None
+
+
+def test_truncate_record_uses_the_given_measure():
+    record = _record_with_results([100])
+    sizes: list[Any] = []
+
+    def measure(value: Any) -> int:
+        sizes.append(value)
+        return 10**6
+
+    out = truncate_record(record, 1000, render=lambda r: r, measure=measure)
+    assert sizes and out["truncated"] is True
+    # The default measure would keep this small record unchanged.
+    assert truncate_record(record, 1000, render=lambda r: r) is record