feat(insight): add async export scheduling

[wangyb-A]

Summary

• move Workflow Insight rendering, truncation, export, and flush off the checkpoint thread
• coalesce cumulative snapshots with one lazy daemon lane per exporter
• drain final records and flush exporters with one shared configurable timeout
• isolate exporter failures and bound pending state without failing workflows

Tracks #687.

Design

• one lazy daemon worker per exporter
• one in-flight record and latest pending record per execution ARN
• FIFO fairness across execution ARNs
• export_timeout_seconds defaults to 5 seconds and bounds drain plus flush
• workers stop when idle; blocked workers are retained and never replaced
• no core SDK changes

Validation

• 113 Insight tests passed, including local-runner e2e
• scheduler and async subset passed 12 consecutive runs
• mypy passed
• Ruff lint and format passed
• wheel and sdist built
• full repository collection: 3537 tests, no errors
• Workflow Insight conformance PR feat: support CONTEXT operations in CheckpointedResult.create_from_op… #73: CloudWatch 18/18 and S3 18/18 (unchanged; record content, ordering, and timeout are unaffected by these changes)

Review decisions

Iteration	Finding	Decision	Reason	Validation
1	Deepcopy failure could alias records across lanes	Fixed	Violated exporter isolation	Copy-failure and no-alias tests
1	Deepcopy fallback lacked tests	Fixed	Regression would be silent	Focused tests repeated
1	_inflight_arn was dead state	Fixed	Misleading and unused	Full suite and mypy
1	Bounded deque scans are O(n)	Declined	Queue capped at 1024; rewrite adds risk	Cap and fairness tests
1	Warm timeout behavior lacked coverage	Fixed	Required approved behavior	Deterministic A/B FIFO test
2	Global thread-count assertion could flake	Fixed	Daemon exit could alter baseline	Lane-local checks; subset 12x
2	Product concurrency and lifecycle	Accepted	No product defects found	Full reviewer trace
3	Full implementation after fixes	Accepted	No actionable findings	Final reviewer pass
3	Timing assertions under severe host load	Accepted risk	Wide margins; 12 repeats passed	Subset 12x
3	Shared flush can publish another execution buffer early	Accepted by design	Non-lossy shared lifecycle behavior	Warm-container tests
public	Cancelled flush barriers could accumulate behind a blocked exporter	Fixed	A stale barrier per warm invocation grew queue/barrier state without bound	Repeated-timeout test + queued/already-popped cancellation-race tests; scheduler subset 12x; CloudWatch 18/18 and S3 18/18
public	Lane used the default reentrant RLock	Fixed	Replaced with an explicit non-reentrant Lock; the lane never re-acquires _cond while holding it, so recursion support is unneeded and misuse now fails loudly	Full insight suite; mypy; ruff; scheduler subset 12x
public	Same exporter instance could be configured more than once	Fixed	WorkflowInsightConfig now rejects a duplicate exporter instance (by object identity, not equality/hash) with a clear ValueError; preserves one-thread-per-distinct-instance safety and avoids duplicate, timing-dependent scheduling. Distinct same-class instances and the default exporter are unaffected	Tests: same instance twice raises; two distinct instances each get a lane; default exporter unaffected; full insight suite; mypy; ruff
public	_ExecutionState.scheduled was read/written without the plugin lock	Fixed (defensive)	Route the flag through _lock via _mark_scheduled/_was_scheduled, consistent with the other state fields; the lock is released before any scheduler/end_invocation or exporter work, so no new lock ordering or deadlock. SDK serializes hooks, so scope is minimal	Scheduled-flag and no-op-invocation tests; scheduler subset 12x

Reviewed three times with commit-code-reviewer, plus two post-public-review passes addressing the rows above. No actionable findings remain.

[wangyb-A]

/ai review

[wangyb-A]

/ai review

[github-actions]

Codex AI review · Finding arf_v1_fpgb65yygmtxhv5a24c72e2brx

[P2] Bound timeouts to threading.TIMEOUT_MAX. Configuration currently accepts any finite positive float, but Event.wait() raises OverflowError above the platform limit. With a blocked exporter, the plugin exception is swallowed before lane shutdown and execution-state cleanup, silently bypassing the drain and retaining state. Reject or clamp oversized values during validation, handle huge integers that overflow math.isfinite, and add a blocked-exporter test.

[github-actions]

Codex AI review

Found one timeout-validation edge case that can bypass exporter drain and plugin cleanup.

Reviewed commit 2e7049ca286f113cf303d8add7d0ed318f6db1e7. Workflow run

[github-actions]

Claude AI review

Reviewed the full diff for the Workflow Insight async-export feature: the new _export_scheduler.py (lane/worker/flush-barrier scheduler), the plugin.py rewiring of _emit into _schedule_record/_build_record plus the new scheduled gate, and the new WorkflowInsightConfig validation (export_timeout_seconds, exporter-instance identity check), along with the accompanying tests.

Traced the concurrency model in detail:

• _ExporterLane's queue/pending invariants (a _pending entry always has a matching queued _RECORD token unless the arn is in flight) hold under all mutation paths (schedule, cap eviction, coalescing, worker pop), so _move_record_token_to_back/_remove_record_token can't silently no-op in a way that corrupts fairness or drops a record.
• The shared record dict handed to every lane is never mutated before each lane's own copy.deepcopy in _export_one; truncate_record also never mutates its input, so cross-lane isolation holds even when a user render() mutates its argument.
• _ExportScheduler.end_invocation's flush barrier is enqueued after any already-queued record token for that lane, so a successful (non-degraded) end_invocation() return is only reached after the corresponding export has actually run — this preserves the pre-existing synchronous-looking behavior for callers/tests, so the many unchanged tests in test_plugin.py that assert on exporter.records immediately after on_invocation_end() still hold correctly under the new async path.
• The new on_operation_change early-return for non-ON_CHANGE modes is safe because on_invocation_end always re-adopts a fresh operations snapshot before building the terminal record, so skipping the mid-invocation adopt has no effect on the final record.
• _ExecutionState.scheduled is consistently read/written through _mark_scheduled/_was_scheduled under the plugin lock, and the gate correctly avoids touching lanes for no-op invocations.
• Cap eviction, cancelled-barrier cleanup, and worker lifecycle (lazy start, never-replace-a-blocked-worker, idle-stop) all match their documented invariants and are exercised by dedicated tests.

No correctness, determinism, thread-safety, or public-API defects were found in the changed code.

Residual test risk: the scheduler tests rely on wall-clock waits (_wait_until, short timeouts like 0.1–0.3s) and thread-name/count introspection to assert concurrency invariants; the PR description already reports repeated runs to guard against this, but such tests can still be more prone to flaking than deterministic tests under heavy CI/host load. Separately, the new exporter-instance identity check (WorkflowInsightConfig._validate_exporters) only guards against reusing the same exporter object within one config; it does not (and probably need not) guard against the same exporter instance being wired into two independent WorkflowInsightConfig/plugin instances, which would still create two lanes calling into a possibly non-thread-safe exporter concurrently — an unlikely but unguarded configuration.

Reviewed commit 2e7049ca286f113cf303d8add7d0ed318f6db1e7. Workflow run