fix(insight): keep one pending export snapshot per execution

[wangyb-A]

Problem / Motivation

The Workflow Insight export scheduler keeps one pending snapshot for the whole plugin. When two executions are in flight on one plugin instance, execution B's RUNNING snapshot can replace execution A's SUCCEEDED snapshot before the worker claims it. A's drain() then returns as soon as B's record is flushed. A's terminal record is never exported.

Raised as issue 1 in #719 (comment). Issue 2 in the same comment says the on-change coalescing test was weakened to pass when records are lost, and the README misstates when coalescing happens.

Why it matters

In on-complete mode the terminal record is the only record, so this is total data loss for that execution. Lambda runs one invocation per environment, so production is mostly safe. The local test runner (aws-durable-execution-sdk-python-testing) drives independent executions concurrently through one shared workflow_insight(...) instance, and the conformance suite runs on top of it. Both are affected.

What changed

_export_scheduler.py: the pending map is keyed by executionArn. A snapshot only replaces its own execution's entry. The worker exports entries in first-arrival order and honors a flush request only once every pending record is exported, so drain() means everything scheduled so far was delivered. Memory is bounded by executions in flight, the same bound the plugin's _state already has. plugin.py is unchanged.

Coalescing within one execution is unchanged. It is now pinned by tests instead of hidden: one test drains between hooks and asserts every on-change record arrives; one blocks the exporter and asserts a burst collapses to first + latest. The README states the literal condition: snapshots for one execution coalesce whenever the worker has not yet claimed the previous one, including when no export is running, and snapshots from different executions never coalesce.

flowchart LR
  subgraph Before
    A1[exec A schedules SUCCEEDED]:::ctx --> S1[one pending slot]:::removed
    B1[exec B schedules RUNNING]:::ctx --> S1
    S1 --> W1[worker exports B only]:::removed
  end
  subgraph After
    A2[exec A schedules SUCCEEDED]:::ctx --> SA[pending A]:::added
    B2[exec B schedules RUNNING]:::ctx --> SB[pending B]:::added
    SA --> W2[worker exports A, then B, then flush]:::changed
    SB --> W2
  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

Loading

🟩 added · 🟨 changed · 🟥 removed · 🟦 unchanged

Each execution now owns its pending slot, so one execution's snapshot never displaces another's.

Not in this PR, on purpose: a "RUNNING may not supersede a pending terminal" rule, closing execution state in plugin.py, and inline export on worker-start failure. Those were the scope in #726 that drew the two P1 race findings, and none of them is needed for the two issues raised.

Tests

• Insight suite: 74 passed (68 before). Ruff check, ruff format, and mypy clean.
• 6 new tests. The 5 issue-1 tests fail against the previous scheduler (verified by swapping the old source back in); the issue-2 contract test passes on both by design, since it pins behavior rather than a defect.
• test_export_scheduler.py: per-execution keying with a blocked exporter; every pending execution exported before flush; two threads × 50 rounds each deliver every terminal record; same-execution burst coalesces to first + latest.
• test_plugin.py: test_on_change_delivers_every_record_when_drained_between_hooks restores the exact ["RUNNING", "RUNNING", "RUNNING", "SUCCEEDED"] assertion; test_concurrent_executions_on_one_plugin_each_deliver_terminal_record drives two executions through one plugin for 50 rounds in both on-complete and on-change, asserting 50 SUCCEEDED per ARN and empty _state after.
• Concurrency tests repeated 20× locally with no flakes.
• Local commit-code-reviewer pass returned no findings. Residual risk it noted: drain() is a global barrier, so a busy concurrent execution can add latency to another's drain. This is the same over-wait the JS drain() incurs by awaiting the shared inFlight, and does not apply on Lambda.

Manual verification

N/A — the two-thread concurrency tests are the reviewer's probe made deterministic and committed.

Screenshots / video

N/A — no user interface changes.

Supersedes #726.

no linked issue: raised as a review comment on #719; #687 tracks broader Workflow Insight follow-up work

[wangyb-A]

/ai review

[github-actions]

Codex AI review

No actionable findings. Residual risk: the concurrency coverage drives plugin hooks directly rather than exercising concurrent executions through the local runner.

Reviewed commit 14fca8bbbd2bc2c7d30bf115317682cce3c21711. Workflow run