[Bug]: Resumed map or parallel can omit recorded terminal item outcomes

Start with concurrency/executor.py, then trace how operation/map.py and operation/parallel.py call execute() for a STARTED parent. Reproduce the recorded SUCCEEDED, SUCCEEDED, FAILED child state with the deterministic event ordering described in the issue. Done means both APIs retain every terminal child outcome before applying completion policies, and the listed regression cases pass without changing checkpoint formats.

Reference implementation

Merged JavaScript fix: aws/aws-durable-execution-sdk-js#1091 — keep recorded item outcomes when a batch resumes

The JavaScript PR fixes the equivalent defect and includes reconciliation tests for recorded successes, recorded failures, bounded concurrency, unfinished items, and custom completion predicates. Its required behavior and regression cases are the parity reference for this issue. The Python implementation can use a different internal design, but it should produce the same observable result for recorded terminal items.

Expected Behavior

A map or parallel operation can resume while its parent CONTEXT operation is still STARTED. Some nested item or branch contexts can already have terminal SUCCEEDED or FAILED records.

The resumed batch must incorporate every existing terminal child-context outcome before it makes a new completion decision.

For this state:

Operation	Recorded status
Batch parent	STARTED
Item 0	SUCCEEDED
Item 1	SUCCEEDED
Item 2	FAILED
Item 3	STARTED

with:

MapConfig(
    max_concurrency=2,
    completion_config=CompletionConfig(min_successful=2),
)

the resumed BatchResult must satisfy:

[item.status for item in result.all] == [
    BatchItemStatus.SUCCEEDED,
    BatchItemStatus.SUCCEEDED,
    BatchItemStatus.FAILED,
]
result.success_count == 2
result.failure_count == 1
result.has_failure is True
result.status is BatchItemStatus.FAILED
result.completion_reason is CompletionReason.MIN_SUCCESSFUL_REACHED

result.throw_if_error() must raise the recorded item failure.

Item 3 does not need to remain in the result. A previously STARTED item is not a terminal outcome.

Actual Behavior

When the batch parent is STARTED, map_handler() and parallel_handler() call ConcurrentExecutor.execute().

execute() initializes every branch as pending. It does not first identify child contexts that already have terminal records.

The recorded child contexts replay on worker threads. Their events reach the coordinator in thread-completion order. That order can differ from their original completion order and their item-index order.

In the state above:

1. Items 0 and 1 replay their recorded successes.
2. The first success frees a concurrency slot, so item 2 is submitted.
3. The second success reaches min_successful=2.
4. The coordinator completes the batch before item 2's worker reports its already-recorded failure.
5. The final nonblocking event drain does not wait for that event.

The resulting value can be:

[item.status for item in result.all] == [
    BatchItemStatus.SUCCEEDED,
    BatchItemStatus.SUCCEEDED,
    BatchItemStatus.STARTED,
]
result.failure_count == 0
result.has_failure is False
result.status is BatchItemStatus.SUCCEEDED

The execution history still contains item 2's FAILED child-context record.

The resumed parent then checkpoints this incomplete BatchResult. Subsequent replay preserves the incorrect result.

Steps to Reproduce

1. Configure a four-item nested map with:

   MapConfig(
       max_concurrency=2,
       completion_config=CompletionConfig(min_successful=2),
       nesting_type=NestingType.NESTED,
   )
   

2. Produce this first-invocation order:

   1. Items 0 and 1 start.
   2. Item 1 succeeds, which starts item 2.
   3. Item 2 fails and records its terminal child-context failure.
   4. Item 3 starts.
   5. Item 0 succeeds, which reaches min_successful=2.
   6. End the invocation before the batch parent records its terminal checkpoint.

3. Resume with the batch parent still STARTED and child records:

   item 0: SUCCEEDED
   item 1: SUCCEEDED
   item 2: FAILED
   item 3: STARTED
   

4. In a deterministic unit test:

   1. Allow item 0's cached success event to reach the coordinator.
   2. Confirm that the coordinator submits item 2.
   3. Hold item 2 after it observes the recorded failure but before its worker posts the FAILED event.
   4. Allow item 1's cached success event to reach the coordinator.
   5. Observe that the coordinator returns after reaching two successes.

5. Assert that item 2 is reported as STARTED, failure_count is zero, and has_failure is false despite its recorded FAILED context.

The same coordinator code is used by parallel().

SDK Version

Reproduced in:

• aws-durable-execution-sdk-python 2.1.0
• aws-durable-execution-sdk-python 3.0.0 on current main

The relevant executor, map handler, and parallel handler are unchanged between these versions.

Python Version

3.14

Is this a regression?

No.

Additional Context

Relevant Python code:

• concurrency/executor.py
  • Evaluates completion after each worker event.
  • Stops once the current counts meet the completion policy.
  • Performs one nonblocking event drain after that decision.
  • Does not identify terminal child records before scheduling.
• operation/map.py
  • Calls replay() only for a terminal parent.
  • Calls execute() when the parent remains STARTED.
• operation/parallel.py
  • Uses the same distinction.
• concurrency/models.py
  • Calculates has_failure, status, and errors only from items included in the returned BatchResult.

Success criteria

1. Recorded outcomes are retained

   • Every existing terminal SUCCEEDED or FAILED nested child-context record appears with that terminal status in the resumed BatchResult.
   • A terminal record cannot become STARTED or disappear because of worker-event ordering.

2. Completion waits for reconciliation

   • The resumed batch does not resolve until all child contexts that were terminal at invocation start have been incorporated.
   • The order in which their worker events arrive does not change which recorded outcomes appear.

3. Completion policies use the reconciled state

   • min_successful, fail-fast, failure-count tolerance, and failure-percentage tolerance are evaluated against all recorded terminal outcomes before later items are admitted.
   • A supported monotonic should_complete predicate cannot cause a recorded terminal outcome to be omitted.

4. No unnecessary new work starts

   • If the reconciled terminal records already satisfy a completion rule, no item beyond the recorded range starts.
   • Existing operation-ID ordering and max_concurrency guarantees remain intact.

5. Existing replay behavior remains valid

   • A parent that is already terminal continues replaying its serialized BatchResult or CompletionRecord.
   • Existing normal-payload and ReplayChildren histories require no migration.
   • Existing in-flight STARTED executions remain readable without a checkpoint-format change.

6. Tests cover both APIs

   • Add the exact SUCCEEDED, SUCCEEDED, FAILED reproduction for map().
   • Add equivalent coverage for parallel().
   • Cover fail-fast and failure-tolerance decisions.
   • Cover delayed deserialization or delayed worker-event delivery.
   • Assert that has_failure, failure_count, status, and throw_if_error() reflect the recorded failure.
   • Assert that terminal branch bodies follow their existing replay behavior.
   • Assert that later item bodies do not run once the reconciled state has completed the batch.

7. Documentation distinguishes two cases

   • A STARTED parent must reconcile terminal child records before completing.
   • After the parent itself has checkpointed its BatchResult, that result remains authoritative on replay. A child can still appear terminal in history while the recorded batch item remains STARTED, as documented for JavaScript by aws/aws-durable-execution-sdk-js#1092.

Out of scope

• Changing NestingType.FLAT, which has no per-item child-context record to reconcile.
• Changing the accepted result/history mismatch after the batch parent has recorded its result.
• Rejecting new operations under orphaned branches, tracked by #641.
• Invocation deadlines and unbounded branch cleanup, tracked by #741.
• Changing any public method signature or checkpoint payload format.