fix(telemetry): write error shape, plain results and log envelope

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

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

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()}

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