diff --git a/packages/aws-durable-execution-sdk-python-insight/README.md b/packages/aws-durable-execution-sdk-python-insight/README.md new file mode 100644 index 00000000..d93ef40f --- /dev/null +++ b/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). diff --git a/packages/aws-durable-execution-sdk-python-insight/pyproject.toml b/packages/aws-durable-execution-sdk-python-insight/pyproject.toml new file mode 100644 index 00000000..fa0e1cb7 --- /dev/null +++ b/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"] diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/__about__.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/__about__.py new file mode 100644 index 00000000..c7c5adad --- /dev/null +++ b/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" diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/__init__.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/__init__.py new file mode 100644 index 00000000..5eaca776 --- /dev/null +++ b/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", +] diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters.py new file mode 100644 index 00000000..6b10fe7a --- /dev/null +++ b/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 "" diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/operations_index.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/operations_index.py new file mode 100644 index 00000000..e348b4cd --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/operations_index.py @@ -0,0 +1,98 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Name-keyed operation summary index. + +Direct port of the JS ``operations-index.ts`` (``buildOperationsByName`` / +``withOperationsByName``). Point-access exporters (CloudWatch Logs) carry the +name-keyed ``operationsByName`` map instead of the lossless ``operations`` +array. Operations without a name are skipped; a name that occurs more than once +aggregates metrics and DROPS ``result``/``error`` (no single representative +value). Scalar fields (``type``/``subType``/``status``) reflect the most-recently +seen occurrence (the runtime appends newer operations to the end of the array). +""" + +from __future__ import annotations + +from typing import Any + + +def build_operations_by_name( + operations: list[dict[str, Any]], +) -> dict[str, dict[str, Any]]: + groups: dict[str, dict[str, Any]] = {} + + for op in operations: + name = op.get("name") + if not name: + continue + + duration = op.get("durationMs") + duration = duration if isinstance(duration, (int, float)) else None + attempt = op.get("attempt") + attempt = attempt if isinstance(attempt, int) else None + failed = 1 if op.get("status") == "FAILED" else 0 + + existing = groups.get(name) + if existing is None: + summary: dict[str, Any] = { + "type": op.get("type"), + "count": 1, + "failedCount": failed, + "status": op.get("status"), + } + if op.get("subType") is not None: + summary["subType"] = op.get("subType") + if duration is not None: + summary["minDurationMs"] = duration + summary["maxDurationMs"] = duration + summary["totalDurationMs"] = duration + if attempt is not None: + summary["maxAttempt"] = attempt + if op.get("result") is not None: + summary["result"] = op.get("result") + if op.get("error") is not None: + summary["error"] = op.get("error") + groups[name] = summary + continue + + # Repeated name: aggregate and drop the per-occurrence result/error. + existing["count"] += 1 + existing["failedCount"] += failed + existing["type"] = op.get("type") + existing["status"] = op.get("status") + if op.get("subType") is not None: + existing["subType"] = op.get("subType") + else: + existing.pop("subType", None) + if duration is not None: + existing["minDurationMs"] = ( + duration + if existing.get("minDurationMs") is None + else min(existing["minDurationMs"], duration) + ) + existing["maxDurationMs"] = ( + duration + if existing.get("maxDurationMs") is None + else max(existing["maxDurationMs"], duration) + ) + existing["totalDurationMs"] = ( + existing.get("totalDurationMs") or 0 + ) + duration + if attempt is not None: + existing["maxAttempt"] = ( + attempt + if existing.get("maxAttempt") is None + else max(existing["maxAttempt"], attempt) + ) + existing.pop("result", None) + existing.pop("error", None) + + return groups + + +def with_operations_by_name(record: dict[str, Any]) -> dict[str, Any]: + """Return the record with ``operations`` replaced by ``operationsByName``.""" + out = {key: value for key, value in record.items() if key != "operations"} + out["operationsByName"] = build_operations_by_name(record.get("operations", [])) + return out diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/plugin.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/plugin.py new file mode 100644 index 00000000..6ad8ffbf --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/plugin.py @@ -0,0 +1,410 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Workflow Insight instrumentation plugin for the Durable Execution Python SDK. + +Port of the JS ``workflowInsight()`` (``aws-durable-execution-sdk-js-insight/src/ +index.ts``). 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. + +Capability notes vs. the JS plugin (recorded, not hidden): + * The JS hooks carry ``executionInput`` / ``executionResult`` and the full + ``operations`` map on the invocation hooks. The Python SDK's + ``InvocationStartInfo`` / ``InvocationEndInfo`` did not, so this package + ships a minimal SDK extension surfacing ``execution_input`` / + ``execution_result``; the operations map is reconstructed by accumulating + the per-operation ``on_operation_end`` / ``on_operation_change`` hooks into + per-execution state (see ``ExecutionState``). + * The Python SDK has no ``pluginsConfig.childOperationsDepth`` equivalent, so + ``full-tree`` records rely on the child operations being live in the + emitting invocation (true for single-invocation and warm-resume cases). +""" + +from __future__ import annotations + +import datetime +import json +import sys +import threading +from typing import Any, Callable + +from aws_durable_execution_sdk_python.lambda_service import ( + InvocationStatus, + OperationStatus, + OperationType, +) +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationEndInfo, + InvocationStartInfo, + OperationChangeInfo, + OperationEndInfo, + OperationInfo, +) + +from aws_durable_execution_sdk_python_insight.truncation import truncate_record +from aws_durable_execution_sdk_python_insight.types import ( + ContentConfig, + InsightExporter, + OperationOverride, + WorkflowInsightConfig, +) + + +_TERMINAL_OP_STATUSES = frozenset( + { + OperationStatus.SUCCEEDED, + OperationStatus.FAILED, + OperationStatus.TIMED_OUT, + OperationStatus.CANCELLED, + OperationStatus.STOPPED, + } +) + +# 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 +# still in flight, so surface it as RUNNING (mirrors the JS STATUS_MAP). +_STATUS_MAP = { + InvocationStatus.SUCCEEDED: "SUCCEEDED", + InvocationStatus.FAILED: "FAILED", + InvocationStatus.PENDING: "RUNNING", + InvocationStatus.RETRY: "RUNNING", +} + + +def _parse_execution_arn(execution_arn: str) -> dict[str, str]: + # arn::lambda:::function::/durable-execution// + parts = execution_arn.split(":") + last = parts[7] if len(parts) > 7 else "" + segments = last.split("/") + return { + "region": parts[3] if len(parts) > 3 else "", + "accountId": parts[4] if len(parts) > 4 else "", + "functionName": parts[6] if len(parts) > 6 else "", + "qualifier": segments[0] if len(segments) > 0 else "", + "executionName": segments[2] if len(segments) > 2 else "", + "invocationId": segments[3] if len(segments) > 3 else "", + } + + +def _fnv1a32(value: str) -> int: + h = 0x811C9DC5 + for ch in value: + h ^= ord(ch) & 0xFF + h = (h * 0x01000193) & 0xFFFFFFFF + return h + + +def _should_sample(execution_arn: str, rate: float) -> bool: + if rate >= 1: + return True + if rate <= 0: + return False + return _fnv1a32(execution_arn) / 0xFFFFFFFF < rate + + +def _resolve_sampling_rate(rate: float | None) -> float: + if rate is None: + return 1.0 + if not isinstance(rate, (int, float)): + return 1.0 + if rate < 0 or rate > 1: + return max(0.0, min(1.0, float(rate))) + return float(rate) + + +def _iso(ts: Any) -> str | None: + if isinstance(ts, datetime.datetime): + return ts.astimezone(datetime.UTC).isoformat().replace("+00:00", "Z") + return None + + +def _duration_ms(start: Any, end: Any) -> int | None: + if isinstance(start, datetime.datetime) and isinstance(end, datetime.datetime): + return int((end - start).total_seconds() * 1000) + return None + + +def _apply_data_content(value: Any, setting: Any) -> Any: + if setting is False: + return None + if value is None: + return None + if callable(setting): + try: + return setting(value) + except Exception: # noqa: BLE001 - a failing redactor must never leak the raw value + return None + return value + + +def _apply_result_override( + transform: Callable[[Any], Any], raw_result: str | None +) -> Any: + if raw_result is None: + return None + try: + parsed = json.loads(raw_result) + except (json.JSONDecodeError, TypeError): + parsed = raw_result + try: + return transform(parsed) + except Exception: # noqa: BLE001 - untrusted transform must never break emission + return None + + +class _ExecutionState: + __slots__ = ("start_time", "parsed_arn", "sampled_in", "cached_input", "operations") + + def __init__( + self, start_time: Any, parsed_arn: dict[str, str], sampled_in: bool + ) -> None: + self.start_time = start_time + self.parsed_arn = parsed_arn + self.sampled_in = sampled_in + self.cached_input: Any = None + # Insertion-ordered map operation_id -> OperationInfo (creation order, + # since on_operation_start/end fire in order). + self.operations: dict[str, OperationInfo] = {} + + +class WorkflowInsightPlugin(DurableInstrumentationPlugin): + def __init__(self, config: WorkflowInsightConfig) -> None: + self._sampling_rate = _resolve_sampling_rate(config.sampling_rate) + self._emit_mode = config.emit_mode or "on-complete" + self._top_level_only = (config.operation_detail or "top-level") != "full-tree" + content: ContentConfig | None = config.content + self._content = content + ops = content.operations if content and content.operations else None + self._include_errors = ( + True if ops is None or ops.include_errors is None else ops.include_errors + ) + self._overrides_by_name: dict[str, OperationOverride] = {} + if ops is not None: + for override in ops.overrides: + self._overrides_by_name[override.operation_name] = override + self._exporters: list[InsightExporter] = list(config.exporters) + self._state: dict[str, _ExecutionState] = {} + self._lock = threading.Lock() + + # -- state ---------------------------------------------------------------- + + def _get_state(self, execution_arn: str) -> _ExecutionState: + with self._lock: + state = self._state.get(execution_arn) + if state is None: + state = _ExecutionState( + start_time=datetime.datetime.now(datetime.UTC), + parsed_arn=_parse_execution_arn(execution_arn), + sampled_in=_should_sample(execution_arn, self._sampling_rate), + ) + self._state[execution_arn] = state + return state + + def _accumulate(self, execution_arn: str | None, op: OperationInfo) -> None: + if not execution_arn: + return + state = self._get_state(execution_arn) + with self._lock: + existing = state.operations.get(op.operation_id) + # Never downgrade a terminal operation with a later non-terminal event. + if ( + existing is not None + and existing.status in _TERMINAL_OP_STATUSES + and op.status not in _TERMINAL_OP_STATUSES + ): + return + state.operations[op.operation_id] = op + + # -- hooks ---------------------------------------------------------------- + + def on_invocation_start(self, info: InvocationStartInfo) -> None: + if not info.execution_arn: + return + state = self._get_state(info.execution_arn) + if not state.sampled_in: + return + if info.is_first_invocation and info.execution_start_time is not None: + state.start_time = info.execution_start_time + elif info.execution_start_time is not None and state.start_time is None: + state.start_time = info.execution_start_time + state.cached_input = info.execution_input + if self._emit_mode == "on-change": + self._emit( + info.execution_arn, + status="RUNNING", + end_time=None, + output_raw=None, + error=None, + ) + + def on_operation_end(self, info: OperationEndInfo) -> None: + # OperationEndInfo has no execution_arn field; accumulate under every + # tracked execution's state is wrong. It is safe to key by the single + # in-flight execution: the SDK runs one execution per invocation, so the + # most-recently started execution owns this operation. + arn = self._current_execution_arn() + self._accumulate(arn, info) + + def on_operation_change(self, info: OperationChangeInfo) -> None: + for op in info.operations.values(): + self._accumulate(info.execution_arn, op) + + def on_invocation_end(self, info: InvocationEndInfo) -> None: + if not info.execution_arn: + return + state = self._get_state(info.execution_arn) + status = _STATUS_MAP.get(info.status, "RUNNING") + is_terminal = status in ("SUCCEEDED", "FAILED") + is_failure = status == "FAILED" + + if self._emit_mode == "on-change": + should_emit = True + elif self._emit_mode == "on-failure": + should_emit = is_failure + else: # on-complete + should_emit = is_terminal + + if state.sampled_in and should_emit: + self._emit( + info.execution_arn, + status=status, + end_time=datetime.datetime.now(datetime.UTC), + output_raw=info.execution_result, + error=info.error, + ) + + if is_terminal: + with self._lock: + self._state.pop(info.execution_arn, None) + + # -- emission ------------------------------------------------------------- + + def _current_execution_arn(self) -> str | None: + with self._lock: + # The most-recently created state is the in-flight execution. + if not self._state: + return None + return next(reversed(self._state)) + + def _build_operations(self, state: _ExecutionState) -> list[dict[str, Any]]: + records: list[dict[str, Any]] = [] + for op in state.operations.values(): + if op.operation_type == OperationType.EXECUTION: + continue + if not op.name: + continue + if self._top_level_only and op.parent_id: + continue + override = self._overrides_by_name.get(op.name) + if override is not None and override.exclude: + continue + + entry: dict[str, Any] = {"id": op.operation_id, "name": op.name} + entry["type"] = op.operation_type.value + if op.sub_type is not None: + entry["subType"] = op.sub_type.value + if op.parent_id is not None: + entry["parentId"] = op.parent_id + entry["status"] = op.status.value if op.status is not None else "UNKNOWN" + start_iso = _iso(op.start_time) + if start_iso is not None: + entry["startTime"] = start_iso + end_iso = _iso(op.end_time) + if end_iso is not None: + entry["endTime"] = end_iso + dur = _duration_ms(op.start_time, op.end_time) + if dur is not None: + entry["durationMs"] = dur + if op.attempt is not None: + entry["attempt"] = op.attempt + if self._include_errors and op.error is not None: + entry["error"] = {"name": op.error.type, "message": op.error.message} + if override is not None and override.result is not None: + value = _apply_result_override(override.result, op.result) + if value is not None: + entry["result"] = value + records.append(entry) + return records + + def _emit( + self, + execution_arn: str, + *, + status: str, + end_time: Any, + output_raw: str | None, + error: Any, + ) -> None: + state = self._get_state(execution_arn) + arn = state.parsed_arn + start_time = state.start_time + duration = _duration_ms(start_time, end_time) + + content = self._content + record: dict[str, Any] = { + "recordType": "WorkflowInsight", + "schemaVersion": "1.0", + "emittedAt": datetime.datetime.now(datetime.UTC) + .isoformat() + .replace("+00:00", "Z"), + "executionArn": execution_arn, + } + if arn.get("executionName"): + record["executionName"] = arn["executionName"] + record["functionName"] = arn.get("functionName", "") + record["functionQualifier"] = arn.get("qualifier", "") + record["region"] = arn.get("region", "") + record["accountId"] = arn.get("accountId", "") + record["status"] = status + start_iso = _iso(start_time) + if start_iso is not None: + record["startTime"] = start_iso + end_iso = _iso(end_time) + if end_iso is not None: + record["endTime"] = end_iso + if duration is not None: + record["durationMs"] = duration + + parsed_output: Any = None + if output_raw is not None and output_raw != "": + try: + parsed_output = json.loads(output_raw) + except (json.JSONDecodeError, TypeError): + parsed_output = output_raw + input_value = _apply_data_content( + state.cached_input, content.input if content else None + ) + output_value = _apply_data_content( + parsed_output, content.output if content else None + ) + if input_value is not None: + record["input"] = input_value + if output_value is not None: + record["output"] = output_value + if error is not None: + record["error"] = {"name": error.type, "message": error.message} + record["operations"] = self._build_operations(state) + + for exporter in self._exporters: + try: + shaped = truncate_record( + record, exporter.max_record_size_bytes, exporter.render + ) + exporter.export(shaped) + except Exception as exc: # noqa: BLE001 - one exporter must not break others / the execution + # NOTE (parity gap, same as JS Promise.allSettled): exporter + # failures are swallowed so instrumentation never breaks the + # execution. A silently broken exporter is indistinguishable + # from success; we at least log to stderr. + print( + f"[workflow-insight] exporter {type(exporter).__name__} failed: {exc}", + file=sys.stderr, + ) # noqa: T201 + + +def workflow_insight(config: WorkflowInsightConfig) -> WorkflowInsightPlugin: + """Create a Workflow Insight plugin. Mirrors the JS ``workflowInsight()`` factory.""" + return WorkflowInsightPlugin(config) diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/py.typed b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/py.typed new file mode 100644 index 00000000..e69de29b diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/truncation.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/truncation.py new file mode 100644 index 00000000..d826d2e0 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/truncation.py @@ -0,0 +1,107 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Best-effort record size limiter. + +Direct port of the JS ``truncation.ts``. Drop order: + 1. operation ``result`` fields, oldest operation first (each dropped op marked + ``truncated: true``); + 2. whole operations, oldest first (``droppedOperations`` count); + 3. last resort — execution ``input`` then ``output`` (``droppedInput`` / + ``droppedOutput``). + +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. +""" + +from __future__ import annotations + +import json +from typing import Any, Callable + + +def json_byte_size(value: Any) -> int | None: + try: + return len( + json.dumps(value, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + ) + except (TypeError, ValueError): + return None + + +def truncate_record( + record: dict[str, Any], + max_bytes: int | None, + render: Callable[[dict[str, Any]], Any] | None = None, +) -> dict[str, Any]: + render = render or (lambda r: r) + if max_bytes is None or max_bytes <= 0: + return record + + initial = json_byte_size(render(record)) + if initial is None or initial <= max_bytes: + return record + + ops: list[dict[str, Any]] = [dict(op) for op in record.get("operations", [])] + kept = [True] * len(ops) + # Oldest-first by ISO startTime string (UTC 'Z' ISO timestamps sort + # chronologically as strings); operations without a startTime sort last. + order = sorted( + range(len(ops)), key=lambda i: (ops[i].get("startTime") or "\uffff", i) + ) + + any_result = False + dropped_ops = 0 + dropped_input = False + dropped_output = False + + 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_ops > 0: + out["droppedOperations"] = dropped_ops + if dropped_input: + out.pop("input", None) + out["droppedInput"] = True + if dropped_output: + out.pop("output", None) + out["droppedOutput"] = True + return out + + def fits() -> bool: + size = json_byte_size(render(candidate())) + return size is not None and size <= max_bytes + + # Phase 1: drop operation results oldest-first. + for idx in order: + if fits(): + break + if kept[idx] and ops[idx].get("result") is not None: + trimmed = dict(ops[idx]) + trimmed.pop("result", None) + trimmed["truncated"] = True + ops[idx] = trimmed + any_result = True + + # Phase 2: drop whole operations oldest-first. + for idx in order: + if fits(): + break + if kept[idx]: + kept[idx] = False + dropped_ops += 1 + + # Phase 3 (last resort): drop input then output. + if not fits() and record.get("input") is not None: + dropped_input = True + 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: + return record + + return candidate() diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/types.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/types.py new file mode 100644 index 00000000..dd6ef0dc --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/types.py @@ -0,0 +1,88 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Configuration types for the Workflow Insight plugin. + +Mirrors the JS ``WorkflowInsightConfig`` / ``ContentConfig`` / ``OperationOverride`` +(``aws-durable-execution-sdk-js-insight/src/types.ts``). Python uses snake_case +config field names; the *emitted wire record* keeps the JS camelCase field names +(see ``plugin.py``) so records read identically across SDKs. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Callable, Protocol + + +# on-complete (default): emit once at terminal SUCCEEDED/FAILED. +# on-failure: emit once only at terminal FAILED. +# on-change: emit on every operation change and at end (nondeterministic count). +EmitMode = str # "on-complete" | "on-failure" | "on-change" + +# top-level (default): drop any operation with a parentId. +# full-tree: include children of contexts too. +OperationDetail = str # "top-level" | "full-tree" + + +class InsightExporter(Protocol): + """A destination that receives one curated Workflow Insight record. + + ``max_record_size_bytes`` bounds the serialized record body (the plugin's + size limiter measures ``render(record)``); ``None`` disables truncation. + ``render`` maps the canonical record dict to the exact shape the exporter + serializes (identity for array exporters, the ``operationsByName`` expansion + for point-access exporters). + """ + + max_record_size_bytes: int | None + + def render(self, record: dict[str, Any]) -> Any: ... # pragma: no cover + + def export(self, record: dict[str, Any]) -> None: ... # pragma: no cover + + def flush(self) -> None: ... # pragma: no cover + + +@dataclass(frozen=True) +class OperationOverride: + """Per-operation override matched by ``operation_name``. + + ``result`` opts the operation's result into the record via a transform that + receives the checkpointed, JSON-parsed result (the SDK's own serialized form + — the plugin never runs custom Serdes). Mirrors JS ``OperationOverride``. + """ + + operation_name: str + exclude: bool = False + result: Callable[[Any], Any] | None = None + + +@dataclass(frozen=True) +class ContentOperations: + overrides: list[OperationOverride] = field(default_factory=list) + include_errors: bool | None = None + + +@dataclass(frozen=True) +class ContentConfig: + """Controls what data is included in emitted records. + + ``input`` / ``output``: ``False`` omits the field, a callable transforms it, + ``True``/``None`` includes it as-is. Mirrors JS ``ContentConfig``. + """ + + input: bool | Callable[[Any], Any] | None = None + output: bool | Callable[[Any], Any] | None = None + operations: ContentOperations | None = None + + +@dataclass(frozen=True) +class WorkflowInsightConfig: + """Configuration for the Workflow Insight plugin. Mirrors JS ``WorkflowInsightConfig``.""" + + exporters: list[InsightExporter] = field(default_factory=list) + sampling_rate: float | None = None + emit_mode: EmitMode | None = None + operation_detail: OperationDetail | None = None + content: ContentConfig | None = None diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/__init__.py b/packages/aws-durable-execution-sdk-python-insight/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/test_plugin.py b/packages/aws-durable-execution-sdk-python-insight/tests/test_plugin.py new file mode 100644 index 00000000..9e766b7f --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/tests/test_plugin.py @@ -0,0 +1,261 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Unit tests for WorkflowInsightPlugin record building. + +Drives the plugin with the SDK's real hook dataclasses and a capturing exporter +(a test double only at the destination boundary — the plugin logic under test is +exercised end to end, nothing about SDK behavior is mocked). +""" + +from __future__ import annotations + +import datetime +from typing import Any + +from aws_durable_execution_sdk_python.lambda_service import ( + ErrorObject, + InvocationStatus, + OperationStatus, + OperationSubType, + OperationType, +) +from aws_durable_execution_sdk_python.plugin import ( + InvocationEndInfo, + InvocationStartInfo, + OperationEndInfo, +) + +from aws_durable_execution_sdk_python_insight import ( + ContentConfig, + ContentOperations, + OperationOverride, + WorkflowInsightConfig, + workflow_insight, +) + +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) + + +class CaptureExporter: + def __init__(self, max_record_size_bytes: int | None = None, render=None) -> None: + self.max_record_size_bytes = max_record_size_bytes + self._render = render or (lambda r: r) + self.records: list[dict[str, Any]] = [] + + def render(self, record: dict[str, Any]) -> Any: + return self._render(record) + + def export(self, record: dict[str, Any]) -> None: + self.records.append(record) + + def flush(self) -> None: + return None + + +def _step( + name, + status=OperationStatus.SUCCEEDED, + attempt=1, + result=None, + error=None, + parent_id=None, + op_id=None, +): + return OperationEndInfo( + operation_id=op_id or name, + operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name=name, + parent_id=parent_id, + start_time=T0, + is_replayed=False, + status=status, + end_time=T1, + result=result, + error=error, + attempt=attempt, + ) + + +def _run( + plugin, + *, + ops, + status=InvocationStatus.SUCCEEDED, + result='"Hello, World!"', + error=None, + input_value="World", +): + plugin.on_invocation_start( + InvocationStartInfo( + request_id=None, + execution_arn=ARN, + is_first_invocation=True, + execution_start_time=T0, + execution_input=input_value, + ) + ) + for op in ops: + plugin.on_operation_end(op) + plugin.on_invocation_end( + InvocationEndInfo( + request_id=None, + execution_arn=ARN, + is_first_invocation=True, + execution_start_time=T0, + status=status, + error=error, + execution_result=result, + ) + ) + + +def test_basic_success_record(): + exporter = CaptureExporter() + plugin = workflow_insight(WorkflowInsightConfig(exporters=[exporter])) + _run(plugin, ops=[_step("greet")]) + assert len(exporter.records) == 1 + rec = exporter.records[0] + assert rec["recordType"] == "WorkflowInsight" + assert rec["schemaVersion"] == "1.0" + assert rec["executionArn"] == ARN + assert rec["executionName"] == "exec-1" + assert rec["functionName"] == "my-fn" + assert rec["status"] == "SUCCEEDED" + assert rec["input"] == "World" + assert rec["output"] == "Hello, World!" + assert "error" not in rec + assert [op["name"] for op in rec["operations"]] == ["greet"] + op = rec["operations"][0] + assert ( + op["type"] == "STEP" and op["subType"] == "Step" and op["status"] == "SUCCEEDED" + ) + assert op["attempt"] == 1 + assert "result" not in op # results omitted by default + + +def test_on_failure_success_emits_nothing(): + exporter = CaptureExporter() + plugin = workflow_insight( + WorkflowInsightConfig(exporters=[exporter], emit_mode="on-failure") + ) + _run(plugin, ops=[_step("greet")], status=InvocationStatus.SUCCEEDED) + assert exporter.records == [] + + +def test_sampling_zero_emits_nothing(): + exporter = CaptureExporter() + plugin = workflow_insight( + WorkflowInsightConfig(exporters=[exporter], sampling_rate=0) + ) + _run(plugin, ops=[_step("greet")]) + assert exporter.records == [] + + +def test_content_omit_input_output_without_drop_flags(): + exporter = CaptureExporter() + plugin = workflow_insight( + WorkflowInsightConfig( + exporters=[exporter], content=ContentConfig(input=False, output=False) + ) + ) + _run(plugin, ops=[_step("greet")]) + rec = exporter.records[0] + assert "input" not in rec and "output" not in rec + assert "droppedInput" not in rec and "droppedOutput" not in rec + + +def test_result_opt_in(): + exporter = CaptureExporter() + plugin = workflow_insight( + WorkflowInsightConfig( + exporters=[exporter], + content=ContentConfig( + operations=ContentOperations( + overrides=[OperationOverride("compute", result=lambda r: r)] + ) + ), + ) + ) + _run(plugin, ops=[_step("compute", result="42")], result="42") + op = exporter.records[0]["operations"][0] + assert op["result"] == 42 # checkpointed JSON string parsed + + +def test_include_errors_false_drops_op_error_keeps_record_error(): + exporter = CaptureExporter() + plugin = workflow_insight( + WorkflowInsightConfig( + exporters=[exporter], + content=ContentConfig(operations=ContentOperations(include_errors=False)), + ) + ) + err = ErrorObject(message="boom", type="StepError", data=None, stack_trace=None) + op_err = ErrorObject( + message="boom", type="InsightTestError", data=None, stack_trace=None + ) + _run( + plugin, + ops=[_step("failing-step", status=OperationStatus.FAILED, error=op_err)], + status=InvocationStatus.FAILED, + result=None, + error=err, + ) + rec = exporter.records[0] + assert rec["error"]["name"] == "StepError" + assert "error" not in rec["operations"][0] + + +def test_top_level_only_drops_children(): + exporter = CaptureExporter() + plugin = workflow_insight(WorkflowInsightConfig(exporters=[exporter])) + parent = OperationEndInfo( + operation_id="p", + operation_type=OperationType.CONTEXT, + sub_type=OperationSubType.PARALLEL, + name="parallel-work", + parent_id=None, + start_time=T0, + is_replayed=False, + status=OperationStatus.SUCCEEDED, + end_time=T1, + ) + child = _step("branch-a-step", parent_id="p", op_id="c") + _run(plugin, ops=[parent, child]) + names = [op["name"] for op in exporter.records[0]["operations"]] + assert names == ["parallel-work"] + + +def test_full_tree_includes_children_with_parent_id(): + exporter = CaptureExporter() + plugin = workflow_insight( + WorkflowInsightConfig(exporters=[exporter], operation_detail="full-tree") + ) + parent = OperationEndInfo( + operation_id="p", + operation_type=OperationType.CONTEXT, + sub_type=OperationSubType.RUN_IN_CHILD_CONTEXT, + name="parent-context", + parent_id=None, + start_time=T0, + is_replayed=False, + status=OperationStatus.SUCCEEDED, + end_time=T1, + ) + child = _step("child-step", parent_id="p", op_id="c") + _run(plugin, ops=[parent, child]) + ops = {op["name"]: op for op in exporter.records[0]["operations"]} + assert set(ops) == {"parent-context", "child-step"} + assert ops["child-step"]["parentId"] == "p" + + +def test_unnamed_operation_dropped(): + exporter = CaptureExporter() + plugin = workflow_insight(WorkflowInsightConfig(exporters=[exporter])) + unnamed = _step(None, op_id="u") # type: ignore[arg-type] + _run(plugin, ops=[_step("named-step"), unnamed]) + names = [op["name"] for op in exporter.records[0]["operations"]] + assert names == ["named-step"] diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/test_shaping.py b/packages/aws-durable-execution-sdk-python-insight/tests/test_shaping.py new file mode 100644 index 00000000..57b4e27e --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/tests/test_shaping.py @@ -0,0 +1,132 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Unit tests for the pure record-shaping helpers (no AWS).""" + +from __future__ import annotations + +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.truncation import truncate_record + + +def _op(name, **kw): + base = { + "id": kw.get("id", name), + "name": name, + "type": "STEP", + "subType": "Step", + "status": "SUCCEEDED", + } + base.update(kw) + return base + + +def test_by_name_single_occurrence_keeps_result_and_error(): + ops = [_op("greet", result="hi", durationMs=5, attempt=1)] + summary = build_operations_by_name(ops)["greet"] + assert summary["count"] == 1 + assert summary["failedCount"] == 0 + assert summary["result"] == "hi" + assert summary["maxAttempt"] == 1 + + +def test_by_name_repeated_name_drops_result_and_error_and_aggregates(): + ops = [ + _op("task", id="a", result=1, durationMs=2, attempt=1), + _op("task", id="b", result=2, durationMs=4, attempt=2), + _op("task", id="c", result=3, durationMs=6, attempt=1), + ] + summary = build_operations_by_name(ops)["task"] + assert summary["count"] == 3 + assert "result" not in summary + assert "error" not in summary + assert summary["maxAttempt"] == 2 + assert summary["minDurationMs"] == 2 + assert summary["maxDurationMs"] == 6 + assert summary["totalDurationMs"] == 12 + + +def test_by_name_failed_count(): + ops = [ + _op("task", id="a", status="FAILED"), + _op("task", id="b", status="SUCCEEDED"), + ] + summary = build_operations_by_name(ops)["task"] + assert summary["failedCount"] == 1 + assert summary["count"] == 2 + + +def test_unnamed_operations_are_skipped_in_index(): + ops = [_op("named"), {"id": "x", "type": "STEP", "status": "SUCCEEDED"}] + result = build_operations_by_name(ops) + assert set(result.keys()) == {"named"} + + +def test_with_operations_by_name_replaces_array(): + record = {"recordType": "WorkflowInsight", "operations": [_op("greet")]} + shaped = with_operations_by_name(record) + assert "operations" not in shaped + assert "operationsByName" in shaped + assert shaped["operationsByName"]["greet"]["count"] == 1 + + +def _record_with_results(sizes): + ops = [] + for i, size in enumerate(sizes): + ops.append( + { + "id": f"{i:016x}", + "name": f"bulk-{i + 1}", + "type": "STEP", + "subType": "Step", + "status": "SUCCEEDED", + "startTime": f"2026-01-01T00:00:0{i}.000Z", + "attempt": 1, + "result": "x" * size, + } + ) + return { + "recordType": "WorkflowInsight", + "schemaVersion": "1.0", + "executionArn": "arn:aws:lambda:us-west-2:123456789012:function:fn:$LATEST/durable-execution/exec/inv", + "status": "SUCCEEDED", + "startTime": "2026-01-01T00:00:00.000Z", + "input": "World", + "output": "done", + "operations": ops, + } + + +def test_truncation_phase1_drops_results_oldest_first_keeps_all_ops(): + record = _record_with_results([2000, 2000, 2000]) + out = truncate_record(record, 4096, render=lambda r: r) + assert out["truncated"] is True + assert "droppedOperations" not in out + ops = {op["name"]: op for op in out["operations"]} + assert len(ops) == 3 + assert "result" not in ops["bulk-1"] and ops["bulk-1"]["truncated"] is True + assert "result" not in ops["bulk-2"] and ops["bulk-2"]["truncated"] is True + assert "result" in ops["bulk-3"] # newest keeps its result + + +def test_truncation_phase2_drops_whole_ops_oldest_first(): + # bulk-1 and bulk-2 carry oversized results; bulk-3 has none. After Phase 1 + # drops both results the record is still over the limit, forcing Phase 2 to + # drop whole operations oldest-first (mirrors insight-16). + record = _record_with_results([2000, 2000, 0]) + record["operations"][2].pop("result", None) + out = truncate_record(record, 480, render=lambda r: r) + assert out["truncated"] is True + assert out.get("droppedOperations", 0) >= 1 + names = {op["name"] for op in out["operations"]} + assert "bulk-1" not in names # oldest dropped + assert "bulk-3" in names # newest retained + + +def test_truncation_noop_when_within_limit(): + record = _record_with_results([5]) + out = truncate_record(record, 5_000_000, render=lambda r: r) + assert out is record diff --git a/pyproject.toml b/pyproject.toml index 6ea8e7c1..93695e42 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -174,6 +174,7 @@ select = ["E4", "E7", "E9", "F", "TID252"] # pycodestyle (E4/E7/E9) + Pyflakes [tool.ruff.lint.isort] known-first-party = [ "aws_durable_execution_sdk_python", + "aws_durable_execution_sdk_python_insight", "aws_durable_execution_sdk_python_otel", "aws_durable_execution_sdk_python_testing", ]