feat: add Workflow Insight instrumentation plugin

packages/aws-durable-execution-sdk-python-insight/README.md

@@ -0,0 +1,55 @@
+# AWS Durable Execution SDK for Python — Workflow Insight plugin
+
+Workflow Insight instrumentation plugin for the AWS Durable Execution SDK for
+Python. A port of the JavaScript SDK's `workflowInsight()` plugin: it listens to
+the SDK's instrumentation hooks and emits one curated `WorkflowInsight` record
+per execution to the configured exporters. The wire record keeps the JS
+camelCase field names so records read identically across SDKs.
+
+> **Experimental.** Like its JS counterpart, this plugin is experimental and may
+> change or be removed in future releases.
+
+## Install
+
+```bash
+pip install aws-durable-execution-sdk-python-insight
+# with the S3 exporter's local-dev dependency:
+pip install "aws-durable-execution-sdk-python-insight[s3]"
+```
+
+## Usage
+
+```python
+from aws_durable_execution_sdk_python import durable_execution
+from aws_durable_execution_sdk_python_insight import workflow_insight
+from aws_durable_execution_sdk_python_insight.exporters import S3Exporter
+
+@durable_execution(
+    plugins=[
+        workflow_insight(
+            exporters=[S3Exporter(bucket="my-bucket", prefix="workflow-insight/")],
+        )
+    ]
+)
+def handler(event, context):
+    ...
+```
+
+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
+lossless per-occurrence `operations` array, one object per execution
+(upsert-by-execution-name, so re-emission overwrites rather than appends).
+
+Emission behavior, record schema (`recordType: WorkflowInsight`,
+`schemaVersion: "1.0"`), sampling, content configuration (input/output
+omission, `include_errors`, per-operation result opt-in), truncation phases,
+and `top-level` vs `full-tree` operation detail all mirror the JS plugin.
+Behavior is validated cross-SDK by the `insight` conformance suite
+(`aws-durable-execution-conformance-tests-insight`).
+
+## Requirements
+
+- `aws-durable-execution-sdk-python` with the plugin invocation hooks that
+  surface `execution_input` / `execution_result` (included since the version
+  this package declares as its minimum).

packages/aws-durable-execution-sdk-python-insight/pyproject.toml

@@ -0,0 +1,79 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "aws-durable-execution-sdk-python-insight"
+dynamic = ["version"]
+description = 'Workflow Insight instrumentation plugin for the AWS Durable Execution SDK for Python'
+readme = "README.md"
+requires-python = ">=3.11"
+license = "Apache-2.0"
+keywords = ["observability", "workflow-insight", "durable-execution"]
+authors = [{ name = "AWS durable-execution-dev", email = "durable-execution-dev@amazon.com" }]
+classifiers = [
+  "Development Status :: 4 - Beta",
+  "Programming Language :: Python",
+  "Programming Language :: Python :: 3.11",
+  "Programming Language :: Python :: 3.12",
+  "Programming Language :: Python :: 3.13",
+  "Programming Language :: Python :: 3.14",
+  "Programming Language :: Python :: Implementation :: CPython",
+]
+dependencies = [
+  # >=1.8.0: first release carrying the plugin invocation-hook fields
+  # (InvocationInfo.execution_input / InvocationEndInfo.execution_result).
+  "aws-durable-execution-sdk-python>=1.8.0",
+]
+
+[project.optional-dependencies]
+# boto3 is provided by the Lambda runtime; declared as an extra for local dev
+# (e.g. the S3Exporter) without vendoring it into deployments.
+s3 = ["boto3>=1.26.0"]
+
+[project.urls]
+Documentation = "https://github.com/aws/aws-durable-execution-sdk-python#readme"
+Issues = "https://github.com/aws/aws-durable-execution-sdk-python/issues"
+Source = "https://github.com/aws/aws-durable-execution-sdk-python"
+
+[tool.hatch.build.targets.sdist.force-include]
+"../../LICENSE" = "LICENSE"
+"../../NOTICE" = "NOTICE"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/aws_durable_execution_sdk_python_insight"]
+
+[tool.hatch.build.targets.wheel.force-include]
+"../../LICENSE" = "aws_durable_execution_sdk_python_insight/LICENSE"
+"../../NOTICE" = "aws_durable_execution_sdk_python_insight/NOTICE"
+
+[tool.hatch.version]
+path = "src/aws_durable_execution_sdk_python_insight/__about__.py"
+
+[tool.hatch.publish.index]
+disable = true
+
+[tool.coverage.run]
+source_pkgs = ["aws_durable_execution_sdk_python_insight"]
+branch = true
+parallel = true
+omit = ["src/aws_durable_execution_sdk_python_insight/__about__.py"]
+
+[tool.coverage.report]
+exclude_lines = ["no cov", "if __name__ == .__main__.:", "if TYPE_CHECKING:"]
+
+[tool.ruff]
+line-length = 88
+target-version = "py311"
+
+[tool.ruff.lint]
+preview = true
+select = ["E4", "E7", "E9", "F", "TID252"]
+
+[tool.ruff.lint.isort]
+known-first-party = ["aws_durable_execution_sdk_python_insight"]
+force-single-line = false
+lines-after-imports = 2
+
+[tool.ruff.lint.per-file-ignores]
+"tests/**" = ["ARG001", "ARG002", "ARG005", "S101", "PLR2004", "PLR6301", "SIM117", "TRY301"]

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

@@ -0,0 +1,4 @@
+# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates.
+#
+# SPDX-License-Identifier: Apache-2.0
+__version__ = "0.0.1"

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

@@ -0,0 +1,43 @@
+# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates.
+#
+# SPDX-License-Identifier: Apache-2.0
+"""Workflow Insight instrumentation plugin for the AWS Durable Execution Python SDK."""
+
+from aws_durable_execution_sdk_python_insight.__about__ import __version__
+from aws_durable_execution_sdk_python_insight.exporters import (
+    LambdaLogExporter,
+    S3Exporter,
+)
+from aws_durable_execution_sdk_python_insight.operations_index import (
+    build_operations_by_name,
+    with_operations_by_name,
+)
+from aws_durable_execution_sdk_python_insight.plugin import (
+    WorkflowInsightPlugin,
+    workflow_insight,
+)
+from aws_durable_execution_sdk_python_insight.truncation import truncate_record
+from aws_durable_execution_sdk_python_insight.types import (
+    ContentConfig,
+    ContentOperations,
+    InsightExporter,
+    OperationOverride,
+    WorkflowInsightConfig,
+)
+
+
+__all__ = [
+    "__version__",
+    "ContentConfig",
+    "ContentOperations",
+    "InsightExporter",
+    "LambdaLogExporter",
+    "OperationOverride",
+    "S3Exporter",
+    "WorkflowInsightConfig",
+    "WorkflowInsightPlugin",
+    "build_operations_by_name",
+    "truncate_record",
+    "with_operations_by_name",
+    "workflow_insight",
+]

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

@@ -0,0 +1,121 @@
+# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates.
+#
+# SPDX-License-Identifier: Apache-2.0
+"""First-party Workflow Insight exporters.
+
+Ports of the JS ``S3Exporter`` and ``LambdaLogExporter``. Both serialize the
+curated record with JS-compatible compact JSON (no whitespace) so the wire bytes
+match across SDKs. Records are written verbatim — no synthetic emission.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from typing import Any
+
+from aws_durable_execution_sdk_python_insight.operations_index import (
+    with_operations_by_name,
+)
+
+
+def _dumps(value: Any) -> str:
+    return json.dumps(value, separators=(",", ":"), ensure_ascii=False)
+
+
+def _sanitize(value: str) -> str:
+    return re.sub(r"[^a-zA-Z0-9._-]", "_", value)
+
+
+class LambdaLogExporter:
+    """Writes ``operationsByName`` records to the function's own log group via ``print``.
+
+    Port of the JS ``LambdaLogExporter``: ``console.log(JSON.stringify(
+    withOperationsByName(record)))``. Requires no extra IAM. Emits the name-keyed
+    summary map (``OPERATIONS_BY_NAME``).
+    """
+
+    def __init__(self, max_record_size_bytes: int | None = None) -> None:
+        self.max_record_size_bytes = (
+            256_000 if max_record_size_bytes is None else max_record_size_bytes
+        )
+
+    def render(self, record: dict[str, Any]) -> dict[str, Any]:
+        return with_operations_by_name(record)
+
+    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(_dumps(self.render(record)), flush=True)  # noqa: T201
+
+    def flush(self) -> None:
+        return None
+
+
+class S3Exporter:
+    """Writes canonical ``operations``-array records to S3.
+
+    Port of the JS ``S3Exporter``. Each record is a JSON object keyed by
+    execution name, so updates to the same execution overwrite the same object.
+    Emits the lossless ``operations`` array (``OPERATIONS_ARRAY``).
+    """
+
+    def __init__(
+        self,
+        bucket: str,
+        prefix: str = "workflow-insight/",
+        partitioning: str = "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
+        self.max_record_size_bytes = (
+            5_000_000 if max_record_size_bytes is None else max_record_size_bytes
+        )
+        if client is not None:
+            self._client = client
+        else:
+            import boto3  # deferred: boto3 is provided by the Lambda runtime
+
+            self._client = (
+                boto3.client("s3", region_name=region) if region else boto3.client("s3")
+            )
+
+    def render(self, record: dict[str, Any]) -> dict[str, Any]:
+        return record
+
+    def export(self, record: dict[str, Any]) -> None:
+        key = self._build_key(record)
+        self._client.put_object(
+            Bucket=self.bucket,
+            Key=key,
+            Body=_dumps(record).encode("utf-8"),
+            ContentType="application/json",
+        )
+
+    def flush(self) -> None:
+        return None
+
+    def _build_key(self, record: dict[str, Any]) -> str:
+        file_name = (
+            _sanitize(
+                record.get("executionName") or record.get("executionArn") or "record"
+            )
+            + ".json"
+        )
+        return f"{self.prefix}{self._partition(record)}{file_name}"
+
+    def _partition(self, record: dict[str, Any]) -> str:
+        if self.partitioning == "function-name":
+            return f"function={_sanitize(record.get('functionName', ''))}/"
+        if self.partitioning == "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] == "-":
+                return f"year={start[0:4]}/month={start[5:7]}/day={start[8:10]}/"
+            return ""
+        return ""