feat: add Workflow Insight instrumentation plugin by wangyb-A · Pull Request #632 · aws/aws-durable-execution-sdk-python

wangyb-A · 2026-08-12T17:25:22Z

Summary

Adds a Workflow Insight instrumentation plugin as a new package,
packages/aws-durable-execution-sdk-python-insight/ — a port of the JS SDK's
workflowInsight() plugin (aws-durable-execution-sdk-js-insight), treated as the
reference implementation throughout. Experimental, matching the JS plugin's status.

It listens to the SDK's instrumentation hooks and emits one curated WorkflowInsight
record (schemaVersion: "1.0") per execution. The wire record keeps the JS camelCase
field names so records read identically across SDKs and land in the same stores/queries.

Behavior (mirrors the JS plugin)

Exporters: LambdaLogExporter default (one JSON line to the function's log group,
carrying the name-keyed operationsByName summary) and S3Exporter (the lossless
per-occurrence operations array; upsert-by-execution-name; none/date/
function-name partitioning). boto3 is an extra ([s3]) since Lambda provides it.
Emit model: on-complete / on-failure / on-change with export coalescing —
a newer record supersedes a pending one; exports never propagate errors into the
execution.
Sampling: deterministic per-execution ARN hash; all-or-nothing per execution.
Content config: input/output omission or transform (redaction), include_errors
gating operation-level error detail only, per-operation result opt-in with optional
transform.
Truncation: phase 1 drops opted-in results oldest-first, phase 2 drops whole
operations oldest-first, input/output last; per-exporter max_record_size_bytes
measured against the exact shape each exporter emits.
Operation detail: top-level (default; children with parentId suppressed) vs
full-tree; unnamed operations are dropped (JS parity).

Depends on #616 (merged)

The plugin reads InvocationInfo.execution_input / InvocationEndInfo.execution_result
introduced by #616 — the dependency floor is set to >=1.8.0 accordingly (first release
that will carry those hooks). Capability note kept in the module docstring: the operations
map is reconstructed by accumulating per-operation hooks into per-execution state (keyed
by execution ARN to isolate warm-container reuse), since Python hooks carry no
end-of-invocation operations snapshot.

Conformance validation (live, us-west-2)

Validated against the cross-SDK insight conformance suite
(aws/aws-durable-execution-conformance-tests#73, 18 requirements): 18/18 on the s3
sink and 18/18 on the cloudwatch sink. Two known cross-SDK divergences are documented
in that suite rather than patched over here: operation ids pass through the SDK's native
blake2b[:64] format (JS uses MD5[:16]; the suite asserts ids as opaque), and the
per-operation error.name surfaces the customer error class while the record-level error
carries the SDK wrapper name (the suite asserts non-empty).

The suite's Python example handlers land in the conformance repo as a follow-up to #73
once this package is available.

Testing

17 unit tests (hatch run test:all packages/aws-durable-execution-sdk-python-insight/tests/)
covering record shaping, operations indexing, truncation phases, sampling, emit modes,
and exporter rendering
hatch fmt clean; package registered in the root known-first-party
Live conformance runs as above (JS-parity behavior confirmed record-for-record)

Port of the JS SDK's workflowInsight() plugin as a new package, aws-durable-execution-sdk-python-insight: listens to the SDK's instrumentation hooks and emits one curated WorkflowInsight record (schemaVersion 1.0, JS-identical camelCase wire format) per execution through configurable exporters (LambdaLogExporter default with the operationsByName summary; S3Exporter with the per-occurrence operations array). Mirrors the JS emit model: on-complete/on-failure/on-change scheduling with coalescing, ARN-hash sampling, content configuration (input/output omission and transforms, include_errors, per-operation result opt-in), two-phase truncation, top-level vs full-tree operation detail, and unnamed-operation dropping. Uses the invocation-hook execution_input/execution_result fields introduced in #616.

Convert the single exporters.py module into an exporters/ package with one module per destination (lambda_log_exporter, s3_exporter) plus a private _common helper, mirroring the JS package's src/exporters/ layout so the set can grow to full parity (DynamoDB, Firehose, CloudWatch Logs, ...) without a single file accreting every backend's imports. Public import paths are unchanged: 'from ...insight import S3Exporter' and 'from ...insight.exporters import S3Exporter' both still resolve. Adds test_exporters.py covering both exporters (previously untested).

…-plugin

- Seed operation map from InvocationStart/End/OperationChange snapshots instead of reconstructing via per-operation hooks (cold-resume correctness) - on-change mode emits an updated RUNNING record on each change - Drop on_operation_end/_current_execution_arn heuristic; key strictly by execution_arn to prevent cross-execution contamination - Clear per-execution state after every invocation end (bounded, no leak on suspend/retry/sampled-out) - Default to LambdaLogExporter when exporters omitted or empty - Always adopt authoritative execution_start_time on resume - Correct hook enum imports (InvocationStatus/OperationType from plugin) - Register insight tests in root testpaths and mypy type-checks

- 1: wire aws-durable-execution-sdk-python-insight into both the build and publish matrices of pypi-publish.yml; the generic legal-file verifier runs through the build matrix unchanged (LICENSE+NOTICE confirmed in whl+sdist). - 3: in on-change mode, a PENDING/RETRY invocation end maps to RUNNING and now omits endTime/durationMs; only terminal SUCCEEDED/FAILED records carry an end time (plus output/error). - 4: fix the README usage example to import WorkflowInsightConfig and call workflow_insight(WorkflowInsightConfig(exporters=[...])); add a smoke test for the documented call shape. - 5: back EmitMode/OperationDetail with StrEnum (JS-style values); config fields use Literal input typing and __post_init__ normalizes accepted strings to enum members (invalid dynamic strings raise ValueError); export the enums. - 6: add a checked-in tests/e2e local-runner integration test that drives the real durable_execution/PluginExecutor lifecycle through a suspend/resume wait and asserts the terminal record includes the prior step and completed wait. Comment 2 (asynchronous export scheduling) is intentionally deferred; no async queue/worker/coalescing/drain was added.

Sign in to view

+    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 ""


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

Sign in to view

+        # 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,
+            )


Sign in to view

+    # 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


Sign in to view

+def _apply_data_content(value: Any, setting: Any) -> Any:
+    if setting is False:
+        return None
+    if value is None:
+        return None


wangyb-A · 2026-08-27T23:35:19Z

Note to reviewer: I am tracking the AI reviewer's comments in an issue: #687
And will have following PRs to fix them.
Will not include the full change in this PR, for easier review

zhongkechen · 2026-08-27T23:54:14Z

+
+[tool.hatch.build.targets.sdist.force-include]
+"../../LICENSE" = "LICENSE"
+"../../NOTICE" = "NOTICE"


add a plugin entry point here to allow this plugin to be auto loaded. See otel plugin for reference

github-actions · 2026-08-28T00:04:50Z

+        for exporter in self._exporters:
+            try:
+                shaped = truncate_record(
+                    record, exporter.max_record_size_bytes, exporter.render
+                )
+                exporter.export(shaped)


Codex AI review

Exporter I/O runs synchronously on the checkpoint critical path. on_operation_change reaches this loop from the checkpoint thread before synchronous checkpoint waiters are released, so every S3 request and retry delays the durable operation; a slow exporter can cause a Lambda timeout despite exceptions being caught. Dispatch exports through a per-execution worker, coalesce pending records to the newest state, and bounded-flush the terminal record. Add a test using a blocking exporter.

github-actions · 2026-08-28T00:04:51Z

+        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


Codex AI review

Valid output states are conflated with omission. A handler returning None serializes as "null", is parsed back to None, and is then omitted. Null input and identity-transformed null operation results are similarly lost. Also, the core hook uses "" for an out-of-band large result, which is silently omitted without truncated/droppedOutput. Use an explicit missing-value sentinel so JSON null is retained, mark unavailable large output as dropped, and apply the same presence-based checks to operation results and operationsByName. Add null and large-output tests.

github-actions · 2026-08-28T00:04:53Z

Codex AI review

Two findings affect execution isolation and record fidelity. Static review only, as required.

Reviewed commit 953e66f62914d9d61c4f165c743533ff2cec4464. Workflow run

wangyb-A had a problem deploying to ai-pr-review August 12, 2026 17:25 — with GitHub Actions Failure

wangyb-A deployed to ai-pr-review August 25, 2026 22:47 — with GitHub Actions Active

wangyb-A had a problem deploying to ai-pr-review-runtime August 25, 2026 22:52 — with GitHub Actions Error

wangyb-A temporarily deployed to ai-pr-review-runtime August 25, 2026 22:52 — with GitHub Actions Inactive