fix: prevent invalid PENDING suspension after checkpoint completion

[zhongkechen]

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

Issue Link, if available

Closes #370

Description

This fixes a race between checkpoint completion and a context thread waiting for the same operation.

Before this change, ExecutionManager processed a checkpoint response in two separate steps:

1. It published the new operation state to operationStorage.
2. It called onCheckpointComplete on the registered operation, which completed its completion future and reactivated any waiting context thread.

A context thread could run between those steps. In the failing interleaving:

1. The checkpoint thread published a terminal state such as SUCCEEDED to operationStorage.
2. Before the checkpoint thread completed the operation future, the context thread entered waitForOperationCompletion and still observed that future as incomplete.
3. The context thread attached its reactivation callback and deregistered itself so the invocation could suspend.
4. If it was the last active thread, suspension began. However, the suspension check now saw only the already-published terminal operation state, so there were no pending operations left.
5. The invocation returned PENDING even though its checkpoint state contained no pending operation, which the durable service rejected with Cannot return PENDING status with no pending operations.

The fix publishes the checkpoint state and notifies the registered operation while holding the same completion lock used by waitForOperationCompletion. This makes the state transition and completion signal atomic from the waiter perspective:

• If the waiter acquires the lock first, operationStorage remains pending until the waiter has safely deregistered, so suspension has a valid pending operation.
• If the checkpoint thread acquires the lock first, it publishes the terminal state, completes the operation future, and synchronously runs the waiter reactivation callback before the waiter can inspect the future or deregister.

The same lock selection is retained for map and parallel children, which use the parent operation completion lock to serialize branch completion. A deterministic regression test holds that lock while a terminal checkpoint is processed and verifies that the terminal state cannot become visible before operation completion.

Demo/Screenshots

Not applicable.

Checklist

• I have filled out every section of the PR template
• I have thoroughly tested this change

Testing

Unit Tests

Yes. Added a deterministic ExecutionManager regression test for the checkpoint publication/completion race.

Commands run:

mvn -pl sdk -Dtest=ExecutionManagerTest test
mvn -pl sdk test

The SDK suite passed 1,150 tests.

Integration Tests

No new integration test was required because the failing interleaving is covered deterministically at the unit boundary. The complete repository reactor, including sdk-integration-tests, passed with:

mvn test

Examples

No new example was required. Existing example tests passed as part of the full reactor; cloud-only tests remained disabled by default.

[zhongkechen]

This fix was included as part of another PR that never had a chance to be merged: #607

[github-actions]

Codex AI review

No actionable findings. Residual risk remains limited to real-backend timing around concurrent checkpoint completion and suspension; this read-only review did not execute tests.

Reviewed commit 11cb10b93a3a6d66d9d08a7f117a9ac0f865ba37. Workflow run