fix(insight): validate S3 partitioning

Address the S3 partition-validation review comment on PR #632. Add a
public S3Partitioning(StrEnum) (DATE=date, FUNCTION_NAME=function-name,
NONE=none) in the s3_exporter module. The constructor is typed as an
S3Partitioning | Literal[...] union (never bare str) and normalizes input
with S3Partitioning(partitioning), so an invalid dynamic value (e.g.
function_name) raises ValueError at construction instead of silently
falling through to no partitioning. Key building now compares enum members.
Re-export S3Partitioning from the exporters package and top-level package
alongside S3Exporter. Existing API-compatible string inputs are preserved.

Scheduler/flush/queueing/draining behavior is intentionally unchanged.

Author: Alex Wang

packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/__init__.py

@@ -7,6 +7,7 @@
 from aws_durable_execution_sdk_python_insight.exporters import (
     LambdaLogExporter,
     S3Exporter,
+    S3Partitioning,
 )
 from aws_durable_execution_sdk_python_insight.operations_index import (
     build_operations_by_name,
@@ -38,6 +39,7 @@
     "OperationDetail",
     "OperationOverride",
     "S3Exporter",
+    "S3Partitioning",
     "WorkflowInsightConfig",
     "WorkflowInsightPlugin",
     "build_operations_by_name",

packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/__init__.py

@@ -27,10 +27,12 @@
 )
 from aws_durable_execution_sdk_python_insight.exporters.s3_exporter import (
     S3Exporter,
+    S3Partitioning,
 )
 
 
 __all__ = [
     "LambdaLogExporter",
     "S3Exporter",
+    "S3Partitioning",
 ]

packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/s3_exporter.py

@@ -5,14 +5,34 @@
 
 from __future__ import annotations
 
-from typing import Any
+from enum import StrEnum
+from typing import Any, Literal
 
 from aws_durable_execution_sdk_python_insight.exporters._common import (
     compact_dumps,
     sanitize,
 )
 
 
+class S3Partitioning(StrEnum):
+    """S3 key partitioning scheme. The values match the JS S3 exporter's options."""
+
+    # year=YYYY/month=MM/day=DD/ derived from the record startTime (default)
+    DATE = "date"
+    # function=<functionName>/
+    FUNCTION_NAME = "function-name"
+    # no partition prefix
+    NONE = "none"
+
+
+# Accepted string inputs, kept in lockstep with the enum values above. The
+# constructor is typed as this ``Literal`` union (never bare ``str``) so a typoed
+# scheme fails a static type check, while ``S3Partitioning(partitioning)`` in
+# ``__init__`` normalizes any accepted value to the matching enum member and
+# raises ``ValueError`` for an invalid dynamic string.
+S3PartitioningInput = Literal["date", "function-name", "none"]
+
+
 class S3Exporter:
     """Writes canonical ``operations``-array records to S3.
 
@@ -25,14 +45,18 @@ def __init__(
         self,
         bucket: str,
         prefix: str = "workflow-insight/",
-        partitioning: str = "date",
+        partitioning: S3Partitioning | S3PartitioningInput = S3Partitioning.DATE,
         region: str | None = None,
         max_record_size_bytes: int | None = None,
         client: Any = None,
     ) -> None:
         self.bucket = bucket
         self.prefix = prefix
-        self.partitioning = partitioning
+        # ``S3Partitioning(x)`` is idempotent for members, accepts the exact
+        # JS-style strings, and raises ``ValueError`` for an unrecognized dynamic
+        # string so an invalid scheme fails at construction rather than silently
+        # falling through to no partitioning.
+        self.partitioning = S3Partitioning(partitioning)
         self.max_record_size_bytes = (
             5_000_000 if max_record_size_bytes is None else max_record_size_bytes
         )
@@ -70,9 +94,9 @@ def _build_key(self, record: dict[str, Any]) -> str:
         return f"{self.prefix}{self._partition(record)}{file_name}"
 
     def _partition(self, record: dict[str, Any]) -> str:
-        if self.partitioning == "function-name":
+        if self.partitioning == S3Partitioning.FUNCTION_NAME:
             return f"function={sanitize(record.get('functionName', ''))}/"
-        if self.partitioning == "date":
+        if self.partitioning == S3Partitioning.DATE:
             start = str(record.get("startTime", ""))
             # YYYY-MM-DD... -> year=YYYY/month=MM/day=DD/
             if len(start) >= 10 and start[4] == "-" and start[7] == "-":

packages/aws-durable-execution-sdk-python-insight/tests/test_exporters.py

@@ -13,13 +13,25 @@
 import json
 from typing import Any
 
-from aws_durable_execution_sdk_python_insight import LambdaLogExporter, S3Exporter
+import pytest
+
+from aws_durable_execution_sdk_python_insight import (
+    LambdaLogExporter,
+    S3Exporter,
+    S3Partitioning,
+)
 from aws_durable_execution_sdk_python_insight.exporters import (
     LambdaLogExporter as LambdaLogExporterFromPkg,
 )
+from aws_durable_execution_sdk_python_insight.exporters import (
+    S3Partitioning as S3PartitioningFromPkg,
+)
 from aws_durable_execution_sdk_python_insight.exporters.s3_exporter import (
     S3Exporter as S3ExporterFromModule,
 )
+from aws_durable_execution_sdk_python_insight.exporters.s3_exporter import (
+    S3Partitioning as S3PartitioningFromModule,
+)
 
 
 def _record(**kw: Any) -> dict[str, Any]:
@@ -59,6 +71,8 @@ def put_object(self, **kwargs: Any) -> None:
 def test_public_import_paths_resolve_same_classes():
     assert LambdaLogExporter is LambdaLogExporterFromPkg
     assert S3Exporter is S3ExporterFromModule
+    assert S3Partitioning is S3PartitioningFromPkg
+    assert S3Partitioning is S3PartitioningFromModule
 
 
 # -- LambdaLogExporter --------------------------------------------------------
@@ -133,3 +147,72 @@ def test_s3_key_falls_back_to_arn_then_record():
     # falls back to the (sanitized) executionArn
     assert client.puts[0]["Key"].endswith(".json")
     assert "arn_aws_lambda" in client.puts[0]["Key"]
+
+
+# -- S3Partitioning -----------------------------------------------------------
+
+
+def test_s3_partitioning_enum_values():
+    # values are the exact JS-compatible strings, and StrEnum members compare
+    # equal to those strings
+    assert S3Partitioning.DATE == "date"
+    assert S3Partitioning.FUNCTION_NAME == "function-name"
+    assert S3Partitioning.NONE == "none"
+    assert {p.value for p in S3Partitioning} == {"date", "function-name", "none"}
+
+
+def test_s3_partitioning_accepts_enum_member_input():
+    client = FakeS3Client()
+    exporter = S3Exporter(
+        bucket="b", partitioning=S3Partitioning.FUNCTION_NAME, client=client
+    )
+    assert exporter.partitioning is S3Partitioning.FUNCTION_NAME
+    exporter.export(_record())
+    assert client.puts[0]["Key"] == "workflow-insight/function=my-fn/exec-1.json"
+
+
+def test_s3_partitioning_normalizes_valid_string_inputs():
+    # existing API-compatible string inputs still work and normalize to members
+    for value, member in (
+        ("date", S3Partitioning.DATE),
+        ("function-name", S3Partitioning.FUNCTION_NAME),
+        ("none", S3Partitioning.NONE),
+    ):
+        exporter = S3Exporter(bucket="b", partitioning=value, client=FakeS3Client())
+        assert exporter.partitioning is member
+
+
+def test_s3_partitioning_default_is_date():
+    exporter = S3Exporter(bucket="b", client=FakeS3Client())
+    assert exporter.partitioning is S3Partitioning.DATE
+
+
+def test_s3_partitioning_date_key_layout():
+    client = FakeS3Client()
+    S3Exporter(bucket="b", partitioning="date", client=client).export(_record())
+    assert (
+        client.puts[0]["Key"]
+        == "workflow-insight/year=2026/month=01/day=01/exec-1.json"
+    )
+
+
+def test_s3_partitioning_function_name_key_layout():
+    client = FakeS3Client()
+    S3Exporter(
+        bucket="b", partitioning="function-name", prefix="wi/", client=client
+    ).export(_record(functionName="fn/weird name", executionName="exec/1"))
+    assert client.puts[0]["Key"] == "wi/function=fn_weird_name/exec_1.json"
+
+
+def test_s3_partitioning_none_key_layout_has_no_prefix_segment():
+    client = FakeS3Client()
+    S3Exporter(bucket="b", partitioning="none", client=client).export(_record())
+    assert client.puts[0]["Key"] == "workflow-insight/exec-1.json"
+
+
+def test_s3_partitioning_invalid_string_raises_value_error():
+    # a dynamic invalid value (e.g. underscore variant) fails at construction
+    with pytest.raises(ValueError):
+        S3Exporter(bucket="b", partitioning="function_name", client=FakeS3Client())
+    with pytest.raises(ValueError):
+        S3Exporter(bucket="b", partitioning="bogus", client=FakeS3Client())