diff --git a/.github/scripts/type-checks.sh b/.github/scripts/type-checks.sh index 8be2ec5f..fa1ce87d 100644 --- a/.github/scripts/type-checks.sh +++ b/.github/scripts/type-checks.sh @@ -13,6 +13,10 @@ mypy --install-types --non-interactive \ packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel \ packages/aws-durable-execution-sdk-python-otel/tests +mypy --install-types --non-interactive \ + packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight \ + packages/aws-durable-execution-sdk-python-insight/tests + # comment out this for now as there are many type check errors in this package #mypy --install-types --non-interactive \ # packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing \ diff --git a/.github/workflows/pypi-publish.yml b/.github/workflows/pypi-publish.yml index b8db9f03..e8be6955 100644 --- a/.github/workflows/pypi-publish.yml +++ b/.github/workflows/pypi-publish.yml @@ -28,6 +28,8 @@ jobs: path: packages/aws-durable-execution-sdk-python-otel - name: aws-durable-execution-sdk-python-testing path: packages/aws-durable-execution-sdk-python-testing + - name: aws-durable-execution-sdk-python-insight + path: packages/aws-durable-execution-sdk-python-insight steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 @@ -64,6 +66,7 @@ jobs: - name: aws-durable-execution-sdk-python - name: aws-durable-execution-sdk-python-otel - name: aws-durable-execution-sdk-python-testing + - name: aws-durable-execution-sdk-python-insight permissions: id-token: write 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..44196691 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/README.md @@ -0,0 +1,62 @@ +# 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 ( + WorkflowInsightConfig, + workflow_insight, +) +from aws_durable_execution_sdk_python_insight.exporters import S3Exporter + +@durable_execution( + plugins=[ + workflow_insight( + WorkflowInsightConfig( + 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..0ea38d47 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/__init__.py @@ -0,0 +1,49 @@ +# 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, + S3Partitioning, +) +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, + EmitMode, + InsightExporter, + OperationDetail, + OperationOverride, + WorkflowInsightConfig, +) + + +__all__ = [ + "__version__", + "ContentConfig", + "ContentOperations", + "EmitMode", + "InsightExporter", + "LambdaLogExporter", + "OperationDetail", + "OperationOverride", + "S3Exporter", + "S3Partitioning", + "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/__init__.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/__init__.py new file mode 100644 index 00000000..3d432de2 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/__init__.py @@ -0,0 +1,38 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""First-party Workflow Insight exporters. + +One module per exporter, mirroring the JS package's ``src/exporters/`` layout +(``aws-durable-execution-sdk-js-insight``). Each destination lives in its own +module so the set can grow to the full JS parity surface (S3, CloudWatch Logs, +DynamoDB, Firehose, EventBridge, SQS, OpenSearch, Redshift, Aurora, HTTP, OTel, +file, ...) without any single file accreting every backend's imports and +optional dependencies. + +Concrete exporters are re-exported here so the public import path is stable: +``from aws_durable_execution_sdk_python_insight.exporters import S3Exporter`` +keeps working exactly as before this package was split out of a single module. +Shared serialization helpers live in the private ``_common`` module. + +Both shipped exporters 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 + +from aws_durable_execution_sdk_python_insight.exporters.lambda_log_exporter import ( + LambdaLogExporter, +) +from aws_durable_execution_sdk_python_insight.exporters.s3_exporter import ( + S3Exporter, + S3Partitioning, +) + + +__all__ = [ + "LambdaLogExporter", + "S3Exporter", + "S3Partitioning", +] diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/_common.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/_common.py new file mode 100644 index 00000000..4401fce1 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/_common.py @@ -0,0 +1,29 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Shared serialization helpers for the Workflow Insight exporters. + +Kept private to the ``exporters`` package: every backend needs the same +JS-compatible compact JSON encoding and the same key/file-name sanitizer, so +they live here rather than being duplicated per exporter module. +""" + +from __future__ import annotations + +import json +import re +from typing import Any + + +def compact_dumps(value: Any) -> str: + """Serialize ``value`` as compact JSON (no whitespace, non-ASCII preserved). + + Matches the JS exporters' ``JSON.stringify`` output so the wire bytes are + identical across SDKs. + """ + return json.dumps(value, separators=(",", ":"), ensure_ascii=False) + + +def sanitize(value: str) -> str: + """Replace characters unsafe for object keys / file names with ``_``.""" + return re.sub(r"[^a-zA-Z0-9._-]", "_", value) diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/lambda_log_exporter.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/lambda_log_exporter.py new file mode 100644 index 00000000..62748d24 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/lambda_log_exporter.py @@ -0,0 +1,39 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Lambda log (CloudWatch) Workflow Insight exporter.""" + +from __future__ import annotations + +from typing import Any + +from aws_durable_execution_sdk_python_insight.exporters._common import compact_dumps +from aws_durable_execution_sdk_python_insight.operations_index import ( + with_operations_by_name, +) + + +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: int | None = ( + 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(compact_dumps(self.render(record)), flush=True) # noqa: T201 + + def flush(self) -> None: + return None diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/s3_exporter.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/s3_exporter.py new file mode 100644 index 00000000..2aca47ae --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/s3_exporter.py @@ -0,0 +1,105 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""S3 Workflow Insight exporter.""" + +from __future__ import annotations + +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=/ + 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. + + 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: 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 + # ``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 + ) + 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=compact_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 == S3Partitioning.FUNCTION_NAME: + return f"function={sanitize(record.get('functionName', ''))}/" + 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] == "-": + 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..d5d23be2 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/plugin.py @@ -0,0 +1,449 @@ +# 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. + +Operation-map sourcing: + The Python SDK invocation hooks now carry the full operation map directly: + ``InvocationStartInfo.operations`` (a point-in-time snapshot at invocation + start), ``InvocationEndInfo.operations`` (a fresh snapshot at invocation end), + and ``OperationChangeInfo.operations`` (the full map at the change). Alongside + them the invocation hooks carry ``execution_arn``, ``execution_start_time``, + ``execution_input`` and ``execution_result``. This plugin reads those + snapshots as the authoritative operation state -- it does NOT reconstruct the + map by accumulating per-operation ``on_operation_end`` events. Because every + invocation start re-seeds the map from the snapshot, a cold resume in a fresh + Lambda environment (a brand-new plugin instance) still reports the prior + terminal operations. + + The Python SDK has no ``pluginsConfig.childOperationsDepth`` equivalent, so + ``full-tree`` records rely on the child operations being present in the + emitting invocation's snapshot (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.plugin import ( + DurableInstrumentationPlugin, + InvocationEndInfo, + InvocationStartInfo, + InvocationStatus, + OperationChangeInfo, + OperationInfo, + OperationType, +) + +from aws_durable_execution_sdk_python_insight.exporters.lambda_log_exporter import ( + LambdaLogExporter, +) +from aws_durable_execution_sdk_python_insight.truncation import truncate_record +from aws_durable_execution_sdk_python_insight.types import ( + ContentConfig, + EmitMode, + InsightExporter, + OperationDetail, + OperationOverride, + WorkflowInsightConfig, +) + + +# 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: dict[InvocationStatus, str] = { + 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", "cached_input", "operations") + + def __init__(self, start_time: Any, parsed_arn: dict[str, str]) -> None: + self.start_time = start_time + self.parsed_arn = parsed_arn + self.cached_input: Any = None + # operation_id -> OperationInfo, adopted verbatim from the SDK's + # authoritative snapshot (invocation start/end and operation-change). + self.operations: dict[str, OperationInfo] = {} + + +class WorkflowInsightPlugin(DurableInstrumentationPlugin): + def __init__(self, config: WorkflowInsightConfig) -> None: + self._sampling_rate = _resolve_sampling_rate(config.sampling_rate) + # config.emit_mode / operation_detail are already normalized to enum + # members (or None) by WorkflowInsightConfig.__post_init__; re-wrap to + # satisfy the static type of the union-typed config fields. + self._emit_mode: EmitMode = ( + EmitMode(config.emit_mode) + if config.emit_mode is not None + else EmitMode.ON_COMPLETE + ) + detail = ( + OperationDetail(config.operation_detail) + if config.operation_detail is not None + else OperationDetail.TOP_LEVEL + ) + self._top_level_only = detail != OperationDetail.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 + # Default-exporter parity with the JS plugin: an omitted OR an explicitly + # empty exporter list falls back to the Lambda log exporter, so the + # plugin is never a silent no-op. A non-empty list is used verbatim. + self._exporters: list[InsightExporter] = ( + list(config.exporters) if config.exporters else [LambdaLogExporter()] + ) + self._state: dict[str, _ExecutionState] = {} + self._lock = threading.Lock() + + # -- sampling / state ----------------------------------------------------- + + def _sampled_in(self, execution_arn: str) -> bool: + # Deterministic per-ARN, so every hook for one execution agrees without + # needing to persist the decision in state. + return _should_sample(execution_arn, self._sampling_rate) + + def _ensure_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), + ) + self._state[execution_arn] = state + return state + + def _discard_state(self, execution_arn: str) -> None: + with self._lock: + self._state.pop(execution_arn, None) + + def _adopt_operations( + self, state: _ExecutionState, operations: dict[str, OperationInfo] + ) -> None: + # Adopt the authoritative point-in-time snapshot. Copy so plugin state + # never aliases the SDK-owned map, and rebind the attribute so a + # concurrent reader holding the prior reference iterates a stable dict. + with self._lock: + state.operations = dict(operations) + + # -- hooks ---------------------------------------------------------------- + + def on_invocation_start(self, info: InvocationStartInfo) -> None: + arn = info.execution_arn + if not arn or not self._sampled_in(arn): + return + state = self._ensure_state(arn) + # Always adopt the service-provided execution start time when present, + # including a cold resume in a fresh environment (never the resume time, + # which would corrupt duration and the date partition). + if info.execution_start_time is not None: + state.start_time = info.execution_start_time + state.cached_input = info.execution_input + # Seed the operation map from the full snapshot on every invocation. On a + # cold resume this rebuilds prior (terminal) operations that a fresh + # plugin instance never saw via per-operation hooks. + self._adopt_operations(state, info.operations) + if self._emit_mode == EmitMode.ON_CHANGE: + self._emit( + arn, + state, + status="RUNNING", + end_time=None, + output_raw=None, + error=None, + ) + + def on_operation_change(self, info: OperationChangeInfo) -> None: + arn = info.execution_arn + if not arn or not self._sampled_in(arn): + return + state = self._ensure_state(arn) + # Replace state with the full operations snapshot carried by the hook. + self._adopt_operations(state, info.operations) + # on-change mode exports an updated RUNNING record on each change so + # mid-invocation progress is observable, not only at start/end. + if self._emit_mode == EmitMode.ON_CHANGE: + self._emit( + arn, + state, + status="RUNNING", + end_time=None, + output_raw=None, + error=None, + ) + + def on_invocation_end(self, info: InvocationEndInfo) -> None: + arn = info.execution_arn + if not arn: + return + if not self._sampled_in(arn): + # Sampled-out executions process no operations and retain no state. + self._discard_state(arn) + return + state = self._ensure_state(arn) + # Refresh from the fresh end-of-invocation snapshot before emitting so + # the terminal record reflects the final operation map. + self._adopt_operations(state, info.operations) + status = _STATUS_MAP.get(info.status, "RUNNING") + is_terminal = status in ("SUCCEEDED", "FAILED") + is_failure = status == "FAILED" + + if self._emit_mode == EmitMode.ON_CHANGE: + should_emit = True + elif self._emit_mode == EmitMode.ON_FAILURE: + should_emit = is_failure + else: # on-complete + should_emit = is_terminal + + if should_emit: + # Only terminal (SUCCEEDED/FAILED) records carry an end time; a + # PENDING/RETRY invocation end maps to RUNNING (still in flight) and + # must omit endTime/durationMs. Passing end_time=None makes _emit + # drop both fields. Output and error likewise belong only to a + # terminal record. + self._emit( + arn, + state, + status=status, + end_time=datetime.datetime.now(datetime.UTC) if is_terminal else None, + output_raw=info.execution_result if is_terminal else None, + error=info.error if is_terminal else None, + ) + + # Clear state after EVERY invocation end, including PENDING/RETRY. The + # next invocation rebuilds it from InvocationStartInfo.operations, so a + # suspended execution that never resumes in this environment (or that was + # sampled out) leaks nothing and state stays bounded. + self._discard_state(arn) + + # -- emission ------------------------------------------------------------- + + def _build_operations( + self, operations: dict[str, OperationInfo] + ) -> list[dict[str, Any]]: + records: list[dict[str, Any]] = [] + for op in 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, + state: _ExecutionState, + *, + status: str, + end_time: Any, + output_raw: str | None, + error: Any, + ) -> None: + arn = state.parsed_arn + start_time = state.start_time + duration = _duration_ms(start_time, end_time) + # Snapshot the operations reference once so a concurrent adopt() rebind + # cannot change the map mid-build. + operations = state.operations + + 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(operations) + + 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..82426609 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/types.py @@ -0,0 +1,121 @@ +# 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 enum import StrEnum +from typing import Any, Callable, Literal, Protocol + + +class EmitMode(StrEnum): + """When the plugin emits a record. The values match the JS plugin's modes.""" + + # emit once at terminal SUCCEEDED/FAILED (default) + ON_COMPLETE = "on-complete" + # emit once only at terminal FAILED + ON_FAILURE = "on-failure" + # emit on every operation change and at end (nondeterministic count) + ON_CHANGE = "on-change" + + +class OperationDetail(StrEnum): + """How much of the operation tree a record carries. Values match the JS plugin.""" + + # drop any operation with a parentId (default) + TOP_LEVEL = "top-level" + # include children of contexts too + FULL_TREE = "full-tree" + + +# Accepted string inputs, kept in lockstep with the enum values above. Config +# fields are typed as these ``Literal`` unions (never bare ``str``) so a typoed +# mode fails a static type check, while ``WorkflowInsightConfig.__post_init__`` +# normalizes any accepted value to the matching enum member and raises +# ``ValueError`` for an invalid dynamic string. +EmitModeInput = Literal["on-complete", "on-failure", "on-change"] +OperationDetailInput = Literal["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 | EmitModeInput | None = None + operation_detail: OperationDetail | OperationDetailInput | None = None + content: ContentConfig | None = None + + def __post_init__(self) -> None: + # Normalize accepted string inputs to enum members so the plugin always + # compares against ``EmitMode`` / ``OperationDetail`` members. ``EmitMode(x)`` + # is idempotent for members, accepts the exact JS-style strings, and raises + # ``ValueError`` for an unrecognized dynamic string. Frozen dataclass, so use + # ``object.__setattr__`` to rebind the normalized value. + if self.emit_mode is not None: + object.__setattr__(self, "emit_mode", EmitMode(self.emit_mode)) + if self.operation_detail is not None: + object.__setattr__( + self, "operation_detail", OperationDetail(self.operation_detail) + ) 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/e2e/__init__.py b/packages/aws-durable-execution-sdk-python-insight/tests/e2e/__init__.py new file mode 100644 index 00000000..ea56ccd9 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/tests/e2e/__init__.py @@ -0,0 +1,3 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/e2e/wait_suspend_resume_int_test.py b/packages/aws-durable-execution-sdk-python-insight/tests/e2e/wait_suspend_resume_int_test.py new file mode 100644 index 00000000..a52bf168 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/tests/e2e/wait_suspend_resume_int_test.py @@ -0,0 +1,92 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""End-to-end test: the terminal Workflow Insight record captures prior work. + +Drives the plugin through the repository's LOCAL durable runner +(``DurableFunctionTestRunner``) and the real ``@durable_execution`` / +``PluginExecutor`` lifecycle -- not by calling hooks directly. A durable +function runs a named step, then suspends on a named wait, then resumes and +completes. The wait completes while the execution is suspended, so the run +takes two invocations (suspend + resume). On the resuming (terminal) +invocation the plugin emits one ``SUCCEEDED`` record, and that record must +include both the earlier step and the now-completed wait -- proving the plugin +sources operations from the SDK's authoritative invocation snapshots across a +suspend/resume boundary. +""" + +from __future__ import annotations + +from typing import Any + +from aws_durable_execution_sdk_python.config import Duration +from aws_durable_execution_sdk_python.context import DurableContext +from aws_durable_execution_sdk_python.execution import ( + InvocationStatus, + durable_execution, +) + +from aws_durable_execution_sdk_python_insight import ( + WorkflowInsightConfig, + workflow_insight, +) +from aws_durable_execution_sdk_python_testing.runner import ( + DurableFunctionTestResult, + DurableFunctionTestRunner, +) + + +_STEP_NAME = "greet" +_WAIT_NAME = "pause" + + +class _CaptureExporter: + """Records every insight record delivered to it (destination boundary only).""" + + 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]) -> dict[str, Any]: + return record + + def export(self, record: dict[str, Any]) -> None: + self.records.append(record) + + def flush(self) -> None: + return None + + +def _insight_handler(event: Any, context: DurableContext) -> str: # noqa: ARG001 + """A named step, then a wait that suspends the execution, then completion.""" + context.step(lambda _step_ctx: "greeted", name=_STEP_NAME) + context.wait(Duration.from_seconds(1), name=_WAIT_NAME) + return "done" + + +def test_terminal_record_includes_prior_step_and_completed_wait() -> None: + capture = _CaptureExporter() + plugin = workflow_insight(WorkflowInsightConfig(exporters=[capture])) + # Functional form (not the decorator-factory form) so the wrapped handler's + # static type stays a plain 2-arg callable for the runner. + handler = durable_execution(_insight_handler, plugins=[plugin]) + + with DurableFunctionTestRunner(handler=handler, execution_timeout=15) as runner: + result: DurableFunctionTestResult = runner.run(input="{}") + + assert result.status is InvocationStatus.SUCCEEDED + + # on-complete default: nothing is emitted for the suspending invocation, one + # terminal record is emitted on the resuming invocation. + assert len(capture.records) == 1 + record = capture.records[0] + assert record["status"] == "SUCCEEDED" + # Terminal record carries an end time / duration (comment 3). + assert "endTime" in record + assert record["durationMs"] is not None + + ops_by_name = {op["name"]: op for op in record["operations"]} + assert _STEP_NAME in ops_by_name, "prior step missing from terminal record" + assert _WAIT_NAME in ops_by_name, "completed wait missing from terminal record" + assert ops_by_name[_STEP_NAME]["status"] == "SUCCEEDED" + assert ops_by_name[_WAIT_NAME]["status"] == "SUCCEEDED" diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/test_config.py b/packages/aws-durable-execution-sdk-python-insight/tests/test_config.py new file mode 100644 index 00000000..623dbc6a --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/tests/test_config.py @@ -0,0 +1,122 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Config normalization tests for the Workflow Insight plugin. + +Covers the ``StrEnum``-backed ``EmitMode`` / ``OperationDetail`` inputs (comment 5): +enum members and JS-style strings both normalize to enum members, defaults resolve +to the documented behavior, and an invalid dynamic string raises ``ValueError``. +Also carries a smoke test for the exact documented call shape (comment 4). +""" + +from __future__ import annotations + +import pytest + +from aws_durable_execution_sdk_python_insight import ( + EmitMode, + OperationDetail, + WorkflowInsightConfig, + workflow_insight, +) +from aws_durable_execution_sdk_python_insight.exporters import S3Exporter + + +# -- enum values match the JS-style wire strings ----------------------------- + + +def test_emit_mode_values(): + assert EmitMode.ON_COMPLETE == "on-complete" + assert EmitMode.ON_FAILURE == "on-failure" + assert EmitMode.ON_CHANGE == "on-change" + + +def test_operation_detail_values(): + assert OperationDetail.TOP_LEVEL == "top-level" + assert OperationDetail.FULL_TREE == "full-tree" + + +# -- string inputs normalize to enum members -------------------------------- + + +@pytest.mark.parametrize( + ("text", "member"), + [ + ("on-complete", EmitMode.ON_COMPLETE), + ("on-failure", EmitMode.ON_FAILURE), + ("on-change", EmitMode.ON_CHANGE), + ], +) +def test_emit_mode_string_input_normalizes_to_enum(text, member): + config = WorkflowInsightConfig(emit_mode=text) + assert config.emit_mode is member + assert isinstance(config.emit_mode, EmitMode) + + +@pytest.mark.parametrize( + ("text", "member"), + [ + ("top-level", OperationDetail.TOP_LEVEL), + ("full-tree", OperationDetail.FULL_TREE), + ], +) +def test_operation_detail_string_input_normalizes_to_enum(text, member): + config = WorkflowInsightConfig(operation_detail=text) + assert config.operation_detail is member + assert isinstance(config.operation_detail, OperationDetail) + + +# -- enum-member inputs pass through unchanged ------------------------------- + + +def test_enum_member_inputs_pass_through(): + config = WorkflowInsightConfig( + emit_mode=EmitMode.ON_CHANGE, operation_detail=OperationDetail.FULL_TREE + ) + assert config.emit_mode is EmitMode.ON_CHANGE + assert config.operation_detail is OperationDetail.FULL_TREE + + +# -- defaults ---------------------------------------------------------------- + + +def test_defaults_are_none_and_plugin_resolves_them(): + config = WorkflowInsightConfig() + assert config.emit_mode is None + assert config.operation_detail is None + plugin = workflow_insight(config) + # Default emit mode is on-complete; default detail is top-level. + assert plugin._emit_mode is EmitMode.ON_COMPLETE + assert plugin._top_level_only is True + + +def test_full_tree_input_disables_top_level_only(): + plugin = workflow_insight(WorkflowInsightConfig(operation_detail="full-tree")) + assert plugin._top_level_only is False + + +# -- invalid dynamic strings raise ------------------------------------------- + + +def test_invalid_emit_mode_string_raises_value_error(): + with pytest.raises(ValueError): + WorkflowInsightConfig(emit_mode="on-compleat") # typo, dynamic value + + +def test_invalid_operation_detail_string_raises_value_error(): + with pytest.raises(ValueError): + WorkflowInsightConfig(operation_detail="whole-tree") # invalid dynamic value + + +# -- README smoke test: the documented call shape must construct (comment 4) -- + + +def test_readme_usage_call_shape_constructs_plugin(): + # Mirrors the README example: workflow_insight(WorkflowInsightConfig(exporters=[...])). + # A stub S3 client avoids any boto3/network dependency while exercising the + # exact documented call. + exporter = S3Exporter( + bucket="my-bucket", prefix="workflow-insight/", client=object() + ) + plugin = workflow_insight(WorkflowInsightConfig(exporters=[exporter])) + assert plugin._exporters == [exporter] diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/test_exporters.py b/packages/aws-durable-execution-sdk-python-insight/tests/test_exporters.py new file mode 100644 index 00000000..84bada35 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/tests/test_exporters.py @@ -0,0 +1,218 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Unit tests for the first-party exporters (no AWS; a fake S3 client is used). + +These lock the behavior preserved when ``exporters.py`` was split into the +``exporters`` package (one module per destination). Imports are exercised via +both the public re-export path and the concrete submodules. +""" + +from __future__ import annotations + +import json +from typing import Any + +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]: + base = { + "recordType": "WorkflowInsight", + "schemaVersion": "1.0", + "executionArn": "arn:aws:lambda:us-west-2:123456789012:function:my-fn:$LATEST/durable-execution/exec-1/inv-1", + "executionName": "exec-1", + "functionName": "my-fn", + "status": "SUCCEEDED", + "startTime": "2026-01-01T00:00:00.000Z", + "operations": [ + { + "id": "a", + "name": "greet", + "type": "STEP", + "subType": "Step", + "status": "SUCCEEDED", + } + ], + } + base.update(kw) + return base + + +class FakeS3Client: + def __init__(self) -> None: + self.puts: list[dict[str, Any]] = [] + + def put_object(self, **kwargs: Any) -> None: + self.puts.append(kwargs) + + +# -- re-export compatibility -------------------------------------------------- + + +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 -------------------------------------------------------- + + +def test_lambda_log_default_size_and_render_is_operations_by_name(): + exporter = LambdaLogExporter() + assert exporter.max_record_size_bytes == 256_000 + shaped = exporter.render(_record()) + 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() + # single compact JSON line, no whitespace separators + assert ", " not in out and '": ' not in out + parsed = json.loads(out) + assert parsed["recordType"] == "WorkflowInsight" + assert "operationsByName" in parsed + + +def test_lambda_log_custom_size_and_flush_noop(): + exporter = LambdaLogExporter(max_record_size_bytes=1234) + assert exporter.max_record_size_bytes == 1234 + assert exporter.flush() is None + + +# -- S3Exporter --------------------------------------------------------------- + + +def test_s3_render_is_identity_and_body_is_compact(): + client = FakeS3Client() + exporter = S3Exporter(bucket="b", client=client) + rec = _record() + assert exporter.render(rec) is rec + exporter.export(rec) + assert len(client.puts) == 1 + put = client.puts[0] + assert put["Bucket"] == "b" + assert put["ContentType"] == "application/json" + body = put["Body"].decode("utf-8") + assert ", " not in body and '": ' not in body + assert json.loads(body)["executionName"] == "exec-1" + + +def test_s3_default_size_and_date_partition_key(): + client = FakeS3Client() + S3Exporter(bucket="b", client=client).export(_record()) + key = client.puts[0]["Key"] + assert key == "workflow-insight/year=2026/month=01/day=01/exec-1.json" + assert S3Exporter(bucket="b", client=client).max_record_size_bytes == 5_000_000 + + +def test_s3_function_name_partition_and_sanitization(): + client = FakeS3Client() + exporter = S3Exporter( + bucket="b", partitioning="function-name", prefix="wi/", client=client + ) + exporter.export(_record(functionName="fn/weird name", executionName="exec/1")) + key = client.puts[0]["Key"] + assert key == "wi/function=fn_weird_name/exec_1.json" + + +def test_s3_key_falls_back_to_arn_then_record(): + client = FakeS3Client() + exporter = S3Exporter(bucket="b", partitioning="none", client=client) + rec = _record() + del rec["executionName"] + exporter.export(rec) + # 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()) 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..087ed0ca --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-insight/tests/test_plugin.py @@ -0,0 +1,570 @@ +# 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). Operations reach the +plugin the way the real SDK delivers them: as the point-in-time ``operations`` +map on ``InvocationStartInfo`` / ``InvocationEndInfo`` / ``OperationChangeInfo``. +""" + +from __future__ import annotations + +import datetime +from typing import Any + +from aws_durable_execution_sdk_python.lambda_service import ( + ErrorObject, + OperationStatus, + OperationSubType, +) +from aws_durable_execution_sdk_python.plugin import ( + InvocationEndInfo, + InvocationStartInfo, + InvocationStatus, + OperationChangeInfo, + OperationEndInfo, + OperationInfo, + OperationType, +) + +from aws_durable_execution_sdk_python_insight import ( + ContentConfig, + ContentOperations, + LambdaLogExporter, + OperationOverride, + WorkflowInsightConfig, + workflow_insight, +) + +ARN = "arn:aws:lambda:us-west-2:123456789012:function:my-fn:$LATEST/durable-execution/exec-1/inv-1" +ARN_B = "arn:aws:lambda:us-west-2:123456789012:function:my-fn:$LATEST/durable-execution/exec-2/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, + op_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + end_time=T1, +) -> OperationInfo: + return OperationEndInfo( + operation_id=op_id or name, + operation_type=op_type, + sub_type=sub_type, + name=name, + parent_id=parent_id, + start_time=T0, + is_replayed=False, + status=status, + end_time=end_time, + result=result, + error=error, + attempt=attempt, + ) + + +def _ops(*ops: OperationInfo) -> dict[str, OperationInfo]: + return {op.operation_id: op for op in ops} + + +def _start( + arn=ARN, + *, + operations: dict[str, OperationInfo] | None = None, + is_first=True, + input_value="World", + execution_start_time=T0, +) -> InvocationStartInfo: + return InvocationStartInfo( + request_id=None, + execution_arn=arn, + is_first_invocation=is_first, + execution_start_time=execution_start_time, + execution_input=input_value, + operations=operations or {}, + ) + + +def _end( + arn=ARN, + *, + operations: dict[str, OperationInfo] | None = None, + status=InvocationStatus.SUCCEEDED, + result='"Hello, World!"', + error=None, + is_first=True, + execution_start_time=T0, +) -> InvocationEndInfo: + return InvocationEndInfo( + request_id=None, + execution_arn=arn, + is_first_invocation=is_first, + execution_start_time=execution_start_time, + status=status, + error=error, + execution_result=result, + operations=operations or {}, + ) + + +def _run( + plugin, + *, + ops, + status=InvocationStatus.SUCCEEDED, + result='"Hello, World!"', + error=None, + input_value="World", +): + """Single-invocation drive: the full operation map is present in both the + start and the end snapshot (the terminal record is built from the end one).""" + operations = _ops(*ops) + plugin.on_invocation_start(_start(operations=operations, input_value=input_value)) + plugin.on_invocation_end( + _end(operations=operations, status=status, result=result, error=error) + ) + + +# -- existing record-building coverage --------------------------------------- + + +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 = _step( + "parallel-work", + op_id="p", + op_type=OperationType.CONTEXT, + sub_type=OperationSubType.PARALLEL, + ) + 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 = _step( + "parent-context", + op_id="p", + op_type=OperationType.CONTEXT, + sub_type=OperationSubType.RUN_IN_CHILD_CONTEXT, + ) + 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"] + + +# -- cold resume: seed from the invocation snapshot (comment 1 + 7) ---------- + + +def test_cold_resume_reports_prior_terminal_ops_with_fresh_plugin(): + # Invocation 1 (plugin A): a step completes, then a wait suspends -> PENDING. + exporter1 = CaptureExporter() + plugin1 = workflow_insight(WorkflowInsightConfig(exporters=[exporter1])) + step = _step("greet", op_id="op-step") + wait_pending = _step( + "pause", + op_id="op-wait", + op_type=OperationType.WAIT, + sub_type=OperationSubType.WAIT, + status=OperationStatus.PENDING, + end_time=None, + ) + plugin1.on_invocation_start(_start(operations={})) + plugin1.on_invocation_end( + _end( + operations=_ops(step, wait_pending), + status=InvocationStatus.PENDING, + result=None, + ) + ) + assert exporter1.records == [] # on-complete emits nothing for a suspend + assert plugin1._state == {} # and retains nothing + + # Invocation 2 on a *fresh* plugin instance (new Lambda environment): the + # resume start snapshot carries the prior terminal step + resolved wait. + exporter2 = CaptureExporter() + plugin2 = workflow_insight(WorkflowInsightConfig(exporters=[exporter2])) + step_done = _step("greet", op_id="op-step") + wait_done = _step( + "pause", + op_id="op-wait", + op_type=OperationType.WAIT, + sub_type=OperationSubType.WAIT, + status=OperationStatus.SUCCEEDED, + ) + resume_ops = _ops(step_done, wait_done) + plugin2.on_invocation_start( + _start(operations=resume_ops, is_first=False, execution_start_time=T0) + ) + plugin2.on_invocation_end( + _end(operations=resume_ops, is_first=False, execution_start_time=T0) + ) + assert len(exporter2.records) == 1 + rec = exporter2.records[0] + names = [op["name"] for op in rec["operations"]] + assert names == ["greet", "pause"] # prior terminal ops present on cold resume + # Comment 7: start time is the original execution start (T0), not the resume + # time -> duration is measured from T0 and is non-negative. + assert rec["startTime"] == "2026-01-01T00:00:00Z" + assert rec["durationMs"] is not None and rec["durationMs"] >= 0 + + +# -- on-change emits an updated record per change (comment 2) ---------------- + + +def test_on_change_emits_running_on_each_change(): + exporter = CaptureExporter() + plugin = workflow_insight( + WorkflowInsightConfig(exporters=[exporter], emit_mode="on-change") + ) + op1 = _step("s1", op_id="1") + op2 = _step("s2", op_id="2") + + plugin.on_invocation_start(_start(operations={})) # RUNNING #1 (start) + plugin.on_operation_change( + OperationChangeInfo( + execution_arn=ARN, updated_operations=_ops(op1), operations=_ops(op1) + ) + ) # RUNNING #2 + plugin.on_operation_change( + OperationChangeInfo( + execution_arn=ARN, updated_operations=_ops(op2), operations=_ops(op1, op2) + ) + ) # RUNNING #3 + plugin.on_invocation_end(_end(operations=_ops(op1, op2))) # SUCCEEDED #4 + + statuses = [r["status"] for r in exporter.records] + assert statuses == ["RUNNING", "RUNNING", "RUNNING", "SUCCEEDED"] + # The record emitted after the 2nd change already carries both operations. + assert [op["name"] for op in exporter.records[2]["operations"]] == ["s1", "s2"] + final = exporter.records[-1] + assert [op["name"] for op in final["operations"]] == ["s1", "s2"] + # No duplicate operation entries within a record (no end/change double-count). + ids = [op["id"] for op in final["operations"]] + assert len(ids) == len(set(ids)) + + +# -- no cross-execution contamination (comment 3) ---------------------------- + + +def test_concurrent_executions_do_not_cross_contaminate(): + # A and B both suspend; B is the most-recently started (the old insertion- + # order heuristic would have attributed A's resume to B). A then resumes to + # a terminal state. Its record must contain only A's data. + exporter = CaptureExporter() + plugin = workflow_insight(WorkflowInsightConfig(exporters=[exporter])) + a_op = _step("a-step", op_id="a1") + b_op = _step("b-step", op_id="b1") + + plugin.on_invocation_start(_start(arn=ARN, operations={}, input_value="A")) + plugin.on_invocation_start(_start(arn=ARN_B, operations={}, input_value="B")) + plugin.on_invocation_end( + _end( + arn=ARN_B, + operations=_ops(b_op), + status=InvocationStatus.PENDING, + result=None, + ) + ) + plugin.on_invocation_end( + _end( + arn=ARN, + operations=_ops(a_op), + status=InvocationStatus.PENDING, + result=None, + ) + ) + assert exporter.records == [] # both suspended, nothing terminal yet + + a_done = _step("a-step", op_id="a1") + plugin.on_invocation_start( + _start(arn=ARN, operations=_ops(a_done), is_first=False, input_value="A") + ) + plugin.on_invocation_end( + _end(arn=ARN, operations=_ops(a_done), is_first=False, result='"A-done"') + ) + + assert len(exporter.records) == 1 + rec = exporter.records[0] + assert rec["executionArn"] == ARN + assert rec["input"] == "A" + assert [op["name"] for op in rec["operations"]] == ["a-step"] + + +# -- state lifecycle: clear after every invocation end (comment 4) ----------- + + +def test_state_cleared_after_pending_and_retry(): + exporter = CaptureExporter() + plugin = workflow_insight(WorkflowInsightConfig(exporters=[exporter])) + op = _step("s", op_id="1") + + plugin.on_invocation_start(_start(operations={})) + plugin.on_invocation_end( + _end(operations=_ops(op), status=InvocationStatus.PENDING, result=None) + ) + assert plugin._state == {} # no leak after suspend + + plugin.on_invocation_start(_start(operations=_ops(op), is_first=False)) + plugin.on_invocation_end( + _end( + operations=_ops(op), + status=InvocationStatus.RETRY, + result=None, + is_first=False, + ) + ) + assert plugin._state == {} # no leak after retry + assert exporter.records == [] # on-complete emits nothing for non-terminal + + +def test_sampled_out_processes_nothing_and_retains_no_state(): + exporter = CaptureExporter() + plugin = workflow_insight( + WorkflowInsightConfig(exporters=[exporter], sampling_rate=0) + ) + op = _step("s", op_id="1") + plugin.on_invocation_start(_start(operations={})) + plugin.on_operation_change( + OperationChangeInfo( + execution_arn=ARN, updated_operations=_ops(op), operations=_ops(op) + ) + ) + plugin.on_invocation_end(_end(operations=_ops(op))) + assert exporter.records == [] + assert plugin._state == {} + + +# -- default exporter parity with JS (comment 6) ----------------------------- + + +def test_default_exporter_when_config_omits_exporters(): + plugin = workflow_insight(WorkflowInsightConfig()) + assert len(plugin._exporters) == 1 + assert isinstance(plugin._exporters[0], LambdaLogExporter) + + +def test_default_exporter_when_exporters_explicitly_empty(): + plugin = workflow_insight(WorkflowInsightConfig(exporters=[])) + assert len(plugin._exporters) == 1 + assert isinstance(plugin._exporters[0], LambdaLogExporter) + + +def test_explicit_exporters_are_preserved(): + exporter = CaptureExporter() + plugin = workflow_insight(WorkflowInsightConfig(exporters=[exporter])) + assert plugin._exporters == [exporter] + + +def test_default_exporter_actually_emits_to_stdout(capsys): + plugin = workflow_insight(WorkflowInsightConfig()) + _run(plugin, ops=[_step("greet")]) + out = capsys.readouterr().out + assert '"recordType":"WorkflowInsight"' in out # compact JSON via LambdaLogExporter + assert '"operationsByName"' in out + + +# -- terminal timestamps: end time only for SUCCEEDED/FAILED (comment 3) ------ + + +def _last_record_for_end_status(status, *, emit_mode="on-change", result=None): + """Drive one invocation to the given end status and return the last record. + + Runs in on-change mode by default so a non-terminal (PENDING/RETRY) end still + emits a RUNNING record whose timestamps we can assert on -- on-complete would + emit nothing for a non-terminal end. + """ + exporter = CaptureExporter() + plugin = workflow_insight( + WorkflowInsightConfig(exporters=[exporter], emit_mode=emit_mode) + ) + op = _step("s", op_id="1") + plugin.on_invocation_start(_start(operations={})) + plugin.on_invocation_end(_end(operations=_ops(op), status=status, result=result)) + return exporter.records[-1] + + +def test_pending_end_maps_to_running_and_omits_end_time(): + rec = _last_record_for_end_status(InvocationStatus.PENDING) + assert rec["status"] == "RUNNING" + assert "endTime" not in rec + assert "durationMs" not in rec + + +def test_retry_end_maps_to_running_and_omits_end_time(): + rec = _last_record_for_end_status(InvocationStatus.RETRY) + assert rec["status"] == "RUNNING" + assert "endTime" not in rec + assert "durationMs" not in rec + + +def test_succeeded_end_is_terminal_with_end_time_and_duration(): + rec = _last_record_for_end_status( + InvocationStatus.SUCCEEDED, result='"Hello, World!"' + ) + assert rec["status"] == "SUCCEEDED" + assert rec["endTime"] is not None + assert rec["durationMs"] is not None + assert rec["output"] == "Hello, World!" + + +def test_failed_end_is_terminal_with_end_time_and_duration(): + err = ErrorObject(message="boom", type="StepError", data=None, stack_trace=None) + exporter = CaptureExporter() + plugin = workflow_insight( + WorkflowInsightConfig(exporters=[exporter], emit_mode="on-change") + ) + op = _step("s", op_id="1", status=OperationStatus.FAILED) + plugin.on_invocation_start(_start(operations={})) + plugin.on_invocation_end( + _end( + operations=_ops(op), + status=InvocationStatus.FAILED, + result=None, + error=err, + ) + ) + rec = exporter.records[-1] + assert rec["status"] == "FAILED" + assert rec["endTime"] is not None + assert rec["durationMs"] is not None + assert rec["error"]["name"] == "StepError" 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 6bc86c8d..2610b41e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -30,6 +30,7 @@ examples-integration = "pytest --runner-mode=cloud -m example packages/aws-durab addopts = "-v --strict-markers --import-mode=importlib" testpaths = [ "packages/aws-durable-execution-sdk-python/tests", + "packages/aws-durable-execution-sdk-python-insight/tests", "packages/aws-durable-execution-sdk-python-otel/tests", "packages/aws-durable-execution-sdk-python-testing/tests", "packages/aws-durable-execution-sdk-python-examples/test", @@ -181,6 +182,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", ]