feat(insight): export records asynchronously

[wangyb-A]

Problem / Motivation

Workflow Insight calls exporters on the SDK checkpoint thread. A slow exporter can delay workflow progress.

Why it matters

Instrumentation must not slow the workflow it observes. The latest terminal snapshot must still drain before normal invocation completion.

What changed

One lazy daemon worker now owns the plugin's exporter list. Checkpoint hooks replace one latest pending snapshot and return. The worker renders, truncates, and exports that snapshot to each exporter. Exporter failures stay isolated.

When an invocation-end hook emits a record, it asks the worker to drain and flush without a timeout, matching the JavaScript implementation. Hooks that emit no record do not start or drain the worker.

flowchart LR
  subgraph Before
    A1[checkpoint hook]:::ctx --> B1[render and export]:::removed --> C1[workflow continues]:::ctx
  end
  subgraph After
    A2[checkpoint hook]:::ctx --> B2[replace latest pending]:::added --> C2[workflow continues]:::ctx
    B2 --> D2[one background worker]:::added --> E2[render export flush]:::changed
  end
  classDef added fill:#DCFCE7,stroke:#16A34A,color:#14532D,stroke-width:2px
  classDef changed fill:#FEF3C7,stroke:#D97706,color:#78350F,stroke-width:2px
  classDef removed fill:#FEE2E2,stroke:#DC2626,color:#7F1D1D,stroke-dasharray:4 3
  classDef ctx fill:#E0F2FE,stroke:#0284C7,color:#0C4A6E
  linkStyle 0,1 stroke:#DC2626,stroke-dasharray:4 3
  linkStyle 2,3,4 stroke:#16A34A,stroke-width:2px

Loading

🟩 added · 🟨 changed · 🟥 removed · 🟦 unchanged

The checkpoint hook now hands the latest snapshot to one worker and returns.

Tests

• Full Workflow Insight suite: 68 passed, including local-runner e2e.
• Focused scheduler/plugin/config suite: 43 passed.
• Latest-pending coalescing, unbounded drain, terminal preservation, idle-hook skip, exporter failure isolation, worker-start failure, and finalizer re-entry tests passed.
• mypy passed.
• Ruff lint and format passed.
• Wheel and sdist built.
• Commit-review pass 2 returned clean at dbaccc9.

Manual verification

N/A — deterministic unit and local-runner coverage exercise the worker lifecycle, unbounded drain, terminal preservation, and idle-hook paths.

Screenshots / video

N/A — no user interface changes.

Supersedes PR #702.

no linked issue: #687 tracks broader Workflow Insight follow-up work

[github-actions]

Codex AI review

One correctness issue remains: a stalled exporter can block invocation completion indefinitely, risking Lambda timeout and replay. Tests do not cover bounded draining.

Reviewed commit dbaccc93964a40d7e2c7905cc6f78ac823573943. Workflow run

[ParidelPooya]

1. Concurrent executions on one plugin lose terminal records

_ExportScheduler._pending is one slot for the whole plugin. It is not keyed by execution ARN. So execution B's RUNNING snapshot can replace execution A's SUCCEEDED snapshot before the worker claims it. A's drain() then waits on the shared
flush event. The worker exports B's record, flushes, sets the event. A's on_invocation_end returns. A's terminal record was never exported.

I measured this with review-probes/probe_concurrent_arns.py: two threads drive two executions through one plugin, 200 trials, count missing SUCCEEDED records.

PR dbaccc9: on-change: 59 lost / 400 on-complete: 73 lost / 400
main: on-change: 0 lost / 400 on-complete: 0 lost / 400

In on-complete mode the terminal record is the only record. So this is an 18% total data-loss rate for concurrent executions, where main has zero.

Two facts make this reachable. First, the plugin already keeps _state: dict[str, _ExecutionState] under a lock, so the plugin itself is designed for several executions in flight. Second, the local runner in
aws-durable-execution-sdk-python-testing runs independent execution lanes concurrently (worker/lane.py), and a module-level workflow_insight(...) instance is shared across every execution in the process. Lambda runs one invocation per
environment, so production is mostly safe. The local runner and the conformance suite are not.

Codex raised the same slot-sharing problem at line 40. The fix in dbaccc9 (remove the timeout) covers the sequential warm-invocation case only. It does not cover two executions active at the same time.

Suggested fix: make _pending a dict[str, dict] keyed by executionArn. The worker drains every entry before it honors a flush request. Memory stays bounded by the number of active executions, which is the same bound _state already has. This is
still far smaller than #702's per-exporter lanes and fixes issue 2 across executions as well.

2. on-change bursts collapse to first + last, and the test was weakened to hide it

This is the finding I left on #702. It is unchanged here.

schedule() sets _pending and calls notify(). The worker exports only when the OS gives it the GIL. If the hook thread calls the next hook before that happens, the new record replaces the old one. review-probes/probe_coalescing.py drives start

• 11 changes + end through one plugin, 20 runs each:

gap between hooks records delivered per run (of 13)
0 ms 2, 2, 2, 2, ... (20/20 runs)
0.5 ms 13 (18 runs), 12 (2 runs)
2 ms 13 (17 runs), 12 (3 runs)

In Lambda a checkpoint round trip separates consecutive hooks, so most records survive there. In the local runner, checkpoints are in-memory and consecutive hooks are microseconds apart. So under the local runner, on-change behaves like
on-complete.

The PR description says this matches JavaScript. It does not. In JS, schedule() calls pump(), and pump() runs synchronously until the first await inside exporter.export(). So JS starts every export immediately and coalesces only while an
export() is truly awaiting I/O. Python coalesces whenever the worker thread has not yet been scheduled. That is a weaker guarantee.

test_on_change_emits_running_on_each_change asserted four statuses ["RUNNING", "RUNNING", "RUNNING", "SUCCEEDED"]. The renamed test asserts statuses[-1] == "SUCCEEDED" and set(statuses[:-1]) <= {"RUNNING"}. Both pass when only one record
arrives. The test cannot detect the loss it was written to guard.

Two deterministic replacements: (a) call plugin._scheduler.drain() after each hook and assert all 13 records arrive, which pins the "no back-pressure means no loss" contract; (b) use a BlockingExporter, schedule three records while blocked,
release, and assert exactly first + last, which pins the coalescing contract. The README sentence "may coalesce while an export is in flight" should state the literal condition: records coalesce whenever the worker has not yet claimed the
previous one, including when no export is running.