[Bug]: Checkpoint batching adds ~100 ms latency to sequential steps

### Expected Behavior

A synchronous checkpoint should be persisted promptly once it is ready. The batcher can combine updates already in the queue, but it should not add an idle collection delay when a caller is blocked waiting for that batch to complete. Callers must still wait for the service acknowledgement and local state refresh before continuing.

### Actual Behavior

With the published Python SDK 1.7.0 and default settings, a loop of sequential steps that each return integer `1` incurs an avoidable approximately 100 ms batch-collection wait per step. The next step cannot start while the current step's synchronous completion checkpoint is waiting for persistence.

Cloud measurements from 2026-09-09:

| Official SDK | Lambda runtime | Successfully checkpointed steps | Lambda invocation duration | Steps/second | Median interval between step completions | p95 interval |
| --- | --- | ---: | ---: | ---: | ---: | ---: |
| Python 1.7.0 | Python 3.13 | 3,000 | 428.882 s | 6.99 | 140 ms | 160 ms |
| Java 2.2.0 | Java 21 | 3,000 | 102.764 s | 29.19 | 32 ms | 53 ms |
| JavaScript 2.4.0 | Node.js 22 | 3,000 | 90.823 s | 33.03 | 29 ms | 39 ms |

All three used `us-east-1`, 1,024 MB, x86_64, a 900-second invocation timeout, a 7,200-second durable execution timeout, default SDK checkpoint/retry behavior, and asynchronous invocation of a published version. Each had a successful three-step warm-up before the measured invocation. The measured invocations were warm, with no initialization duration in their CloudWatch platform reports. This is one measured invocation per SDK; JavaScript ran later on the same day. The comparison does not isolate every source of latency.

**The handlers requested 20,000 steps, but none completed 20,000.** AWS failed each durable execution after exactly 3,000 distinct `StepSucceeded` events with:

```text
ExecutionLimitExceeded:
Execution has exceeded the maximum operations per execution limit of 3000 operations.
```

Every recorded successful step returned `1` on attempt 1. The durations above are actual CloudWatch `platform.report` invocation durations, not projections. All invocations ended before 15 minutes. The service operation cap is separate from this SDK latency issue; a longer durable execution timeout does not remove it. The Service Quotas API returned `L-42D0A120`, value `3000`, `Adjustable: false`.

For context only, linear extrapolation of these measurements to 20,000 steps would be about 47.65 minutes for Python, 11.42 minutes for Java, and 10.09 minutes for JavaScript. These are not measured completion times or guarantees, and the current service cap prevents validating that workload in one execution.

### Steps to Reproduce

1. Package `aws-durable-execution-sdk-python==1.7.0` and `boto3==1.43.90` with the following handler. This is the Python handler used in the cloud experiment:

```python
from aws_durable_execution_sdk_python import DurableContext, durable_execution

@durable_execution
def handler(event: dict, context: DurableContext) -> dict:
count = event.get("steps", 20000)
if not isinstance(count, int) or not 1 <= count <= 20000:
raise ValueError("steps must be an integer between 1 and 20000")

total = 0
for index in range(count):
total += context.step(lambda step_context: 1, name="noop")
if (index + 1) % 1000 == 0 or index + 1 == count:
context.logger.info(
"BENCHMARK_PROGRESS completed=%s total=%s", index + 1, total
)

return {"sdk": "python", "steps": count, "sum": total}
```

2. Deploy with the configuration above, durable execution enabled, JSON logs, and the required Lambda checkpoint/state and logging permissions. Publish a version.
3. Invoke with `{"steps":3}` and confirm the returned sum is `3`.
4. Invoke the same published version asynchronously with `{"steps":20000}` to reproduce the measured run, collecting execution history and CloudWatch logs. To examine latency in an execution that stays below the operation cap, use `{"steps":1000}` instead; the numbers in the table were not measured with that smaller input.
5. Compare timestamps of consecutive `StepSucceeded` events and the Lambda invocation duration. Each loop iteration waits for `context.step()` before starting the next; the step body performs no external I/O or delay.

### SDK Version

Cloud benchmark and local batch-collector probe: `aws-durable-execution-sdk-python==1.7.0`, `boto3==1.43.90`.

The same timed queue-read logic is also present in upstream `main` at `d61985ef697dd9144f986d837475efee91d26a3e`, verified by source inspection. The newer upstream revision was not cloud-benchmarked.

### Python Version

Python 3.13, Lambda `python3.13`.

### Is this a regression?

Unknown. No earlier Python SDK release was benchmarked.

### Last Working Version

Not established.

### Additional Context

#### Checkpoint path and cause

In the measured release, step success calls `state.create_checkpoint(operation_update=success_operation)` with the default `is_sync=True`. The caller blocks on a `CompletionEvent` until the background worker persists the batch and refreshes local execution state.

The collector nevertheless continues trying to add another queue item after it has already collected that synchronous checkpoint:

```python
remaining_time = min(
batch_deadline - time.time(),
0.1,
)
additional_op = self._checkpoint_queue.get(timeout=remaining_time)
```

When no more updates arrive, that read waits approximately 100 ms, raises `queue.Empty`, and only then does the batch flush. For a sequential workflow, the next step's updates cannot arrive until the current synchronous checkpoint finishes, so the wait serves no useful batching purpose.

The initial blocking read while the worker has no work is different: it wakes when an item arrives. The problematic wait is the additional read after a batch already contains an update with a waiting caller.

Source reference: [`ExecutionState._collect_checkpoint_batch()` at the inspected upstream commit](https://github.com/aws/aws-durable-execution-sdk-python/blob/d61985ef697dd9144f986d837475efee91d26a3e/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/state.py#L1020-L1144).

#### Local isolation of the delay

A local five-iteration probe queued an asynchronous START followed by a synchronous SUCCEED, then called the collector directly. No AWS API calls were made:

| Collector | Median batch-collection time |
| --- | ---: |
| Published 1.7.0 collector | 100.225 ms |
| In-memory variant that drains without waiting once the batch contains a synchronous checkpoint | 0.033 ms |

Both collected START and SUCCEED in order and retained the same completion-event object. This only measures batch assembly. The proposed change has not been validated with the full SDK test suite or deployed in a cloud benchmark.

Minimal local probe for the existing delay (no AWS credentials required)

Install `aws-durable-execution-sdk-python==1.7.0`, then run:

```python
import statistics
import time
from unittest.mock import Mock

from aws_durable_execution_sdk_python.lambda_service import (
OperationAction,
OperationType,
OperationUpdate,
)
from aws_durable_execution_sdk_python.plugin import PluginExecutor
from aws_durable_execution_sdk_python.state import ExecutionState, QueuedOperation
from aws_durable_execution_sdk_python.threading import CompletionEvent

samples = []
for _ in range(5):
client = Mock()
state = ExecutionState(
durable_execution_arn="local-probe",
initial_checkpoint_token="local-token",
operations={},
service_client=client,
plugin_executor=PluginExecutor(plugins=None),
)
start = OperationUpdate(
operation_id="step",
operation_type=OperationType.STEP,
action=OperationAction.START,
)
finish = OperationUpdate(
operation_id="step",
operation_type=OperationType.STEP,
action=OperationAction.SUCCEED,
)
waiter = CompletionEvent()
state._checkpoint_queue.put(QueuedOperation(start, None))
state._checkpoint_queue.put(QueuedOperation(finish, waiter))

before = time.perf_counter()
batch = state._collect_checkpoint_batch()
samples.append((time.perf_counter() - before) * 1000)

assert [q.operation_update.action for q in batch] == [
OperationAction.START,
OperationAction.SUCCEED,
]
assert batch[-1].completion_event is waiter
client.checkpoint.assert_not_called()

print(f"Median batch collection: {statistics.median(samples):.3f} ms")
```

#### Proposed change

Make a synchronous checkpoint trigger prompt flushing while preserving batching and persistence semantics:

- Track whether a batch contains a `QueuedOperation` whose `completion_event is not None`, including items taken from the overflow queue and empty checkpoints.
- Once such an item is collected, consume any additional updates already queued using `get_nowait()`, then flush when the queue is empty. Do not keep waiting for future updates while a caller is blocked on the current batch.
- Preserve asynchronous-only batching, FIFO/overflow handling, byte/item limits, and empty-checkpoint coalescing. In particular, allow an asynchronous step START to batch with its synchronous SUCCEED instead of unconditionally flushing START separately.
- Keep checkpoint API calls serialized, maintain checkpoint-token ordering, and wake synchronous callers only after service acknowledgement and local state refresh. Preserve checkpoint failure propagation and shutdown handling.

Conceptually, inside the additional-item collection loop:

```python
if batch_has_sync_checkpoint:
additional_op = self._checkpoint_queue.get_nowait()
else:
additional_op = self._checkpoint_queue.get(timeout=remaining_time)
```

This is a proposed algorithm, not a complete patch; the flag must be maintained for all batch insertion paths.

Suggested validation: prompt flush for synchronous completion checkpoints; asynchronous START + synchronous SUCCEED coalescing; batching under concurrent child contexts; overflow and payload/item limits; empty synchronous checkpoints; failure propagation and shutdown; at-most-once START acknowledgement before user code; successful/failed replay behavior. A regression test should check that the collector does not perform a timed empty-queue wait once it holds a synchronous checkpoint, rather than rely only on a tight wall-clock assertion. Re-run a cloud benchmark below the operation cap after the fix.

Related: #48 introduced checkpoint batching and describes flushing when a non-START checkpoint arrives. This report isolates the idle wait affecting sequential synchronous checkpoints in the current collector.