docs(map,parallel): document completion predicate and fix 2.x semantics

docs/sdk-reference/operations/map.md

@@ -333,8 +333,13 @@ execution and the completion status of the result.
         min_successful: int | None = None
         tolerated_failure_count: int | None = None
         tolerated_failure_percentage: int | float | None = None
+        should_complete: Callable[[CompletionStatus], CompletionDecision] | None = None
     ```
 
+    Use the threshold fields for count-based rules, or `should_complete` for a custom
+    predicate (see [Completion strategies](#completion-strategies)). `should_complete`
+    cannot be combined with the threshold fields.
+
 === "Java"
 
     ```java
@@ -454,7 +459,9 @@ execution and the completion status of the result.
     - **`completion_reason`** why the operation completed. See
         [Completion strategies](#completion-strategies).
     - **`has_failure`** `True` if any item failed
-    - **`throw_if_error()`** raises the first item error as a `CallableRuntimeError`
+    - **`throw_if_error()`** raises the first failure: `ChildContextError` for a failed
+        item, `SerDesError` if an item result failed to serialize, or
+        `BatchCompletionError` if a custom predicate failed the batch
     - **`to_dict()`** serializes to a plain dict. Serializability depends on `R`.
 
 === "Java"
@@ -736,21 +743,42 @@ abandoned items, but cancellation is not guaranteed.
 
 === "Python"
 
-    The `BatchResult`'s `completion_reason` indicates the stop condition. Items that were
-    never started appear in `result.all` with status `STARTED`.
+    The `BatchResult`'s `completion_reason` indicates the stop condition. Items that
+    started but did not complete appear in `result.all` with status `STARTED`. Items that
+    never started are omitted from `result.all`, so `total_count` counts only the items
+    that appear.
 
     | `completion_config`              | Early exit `completion_reason` | Full completion `completion_reason` |
     | -------------------------------- | ------------------------------ | ----------------------------------- |
     | `CompletionConfig()` (default)   | `FAILURE_TOLERANCE_EXCEEDED`   | `ALL_COMPLETED`                     |
     | `first_successful()`             | `MIN_SUCCESSFUL_REACHED`       | `ALL_COMPLETED`                     |
+    | `all_completed()`                | n/a                            | `ALL_COMPLETED`                     |
     | `tolerated_failure_count=N`      | `FAILURE_TOLERANCE_EXCEEDED`   | `ALL_COMPLETED`                     |
     | `tolerated_failure_percentage=N` | `FAILURE_TOLERANCE_EXCEEDED`   | `ALL_COMPLETED`                     |
     | `min_successful=N`               | `MIN_SUCCESSFUL_REACHED`       | `ALL_COMPLETED`                     |
 
-    !!! warning
+    The default `CompletionConfig()` is fail-fast: any item failure exceeds the tolerance
+    and completes the batch early. Use `CompletionConfig.all_completed()` to run every item
+    regardless of failures.
+
+    Set `should_complete` when the threshold fields cannot express the rule. The predicate
+    receives a `CompletionStatus` (`success_count`, `failure_count`, `completed_count`,
+    `total_count`, and `items`, a per-item tuple of `CompletionItemStatus`) and returns a
+    `CompletionDecision`: `continue_batch()` to keep going, or
+    `complete_batch(outcome)` to stop, where `outcome` defaults to
+    `CompletionOutcome.SUCCEEDED`. A `CompletionOutcome.FAILED` outcome marks the whole
+    batch failed, and `throw_if_error()` then raises `BatchCompletionError` even when no
+    individual item failed.
+
+    ```python
+    --8<-- "examples/python/operations/map/custom-completion.py"
+    ```
 
-        `CompletionConfig.all_completed()` is deprecated. Use the default `CompletionConfig()`
-        instead.
+    The predicate runs before any item is scheduled (`completed_count == 0`) and again on
+    each terminal or suspension event, so it must handle the initial zero-progress
+    snapshot. It cannot be combined with `min_successful`, `tolerated_failure_count`, or
+    `tolerated_failure_percentage`, and must be deterministic, side-effect-free, and
+    monotonic. Unscheduled items report `status=None` in `items`.
 
 === "Java"
 

docs/sdk-reference/operations/parallel.md

@@ -324,8 +324,13 @@ execution and the completion status of the result.
         min_successful: int | None = None
         tolerated_failure_count: int | None = None
         tolerated_failure_percentage: int | float | None = None
+        should_complete: Callable[[CompletionStatus], CompletionDecision] | None = None
     ```
 
+    Use the threshold fields for count-based rules, or `should_complete` for a custom
+    predicate (see [Completion strategies](#completion-strategies)). `should_complete`
+    cannot be combined with the threshold fields.
+
 === "Java"
 
     ```java
@@ -462,7 +467,9 @@ execution and the completion status of the result.
     - **`completion_reason`** why the operation completed. See
         [Completion strategies](#completion-strategies).
     - **`has_failure`** `True` if any branch failed
-    - **`throw_if_error()`** raises the first branch error as a `CallableRuntimeError`
+    - **`throw_if_error()`** raises the first failure: `ChildContextError` for a failed
+        branch, `SerDesError` if a branch result failed to serialize, or
+        `BatchCompletionError` if a custom predicate failed the batch
     - **`to_dict()`** serializes to a plain dict. Serializability depends on `R`.
 
     ```python
@@ -763,21 +770,39 @@ ongoing work in abandoned branches, but cancellation is not guaranteed.
 === "Python"
 
     The `BatchResult`'s `completion_reason` indicates the stop condition with which the
-    parallel operation completed. Branches that were never started appear in `result.all`
-    with status `STARTED`.
+    parallel operation completed. Branches that started but did not complete appear in
+    `result.all` with status `STARTED`. Branches that never started are omitted from
+    `result.all`, so `total_count` counts only the branches that appear.
 
     | `completion_config`              | Early exit `completion_reason` | Full completion `completion_reason` |
     | -------------------------------- | ------------------------------ | ----------------------------------- |
     | `all_successful()` (default)     | `FAILURE_TOLERANCE_EXCEEDED`   | `ALL_COMPLETED`                     |
     | `first_successful()`             | `MIN_SUCCESSFUL_REACHED`       | `ALL_COMPLETED`                     |
+    | `all_completed()`                | n/a                            | `ALL_COMPLETED`                     |
     | `tolerated_failure_count=N`      | `FAILURE_TOLERANCE_EXCEEDED`   | `ALL_COMPLETED`                     |
     | `tolerated_failure_percentage=N` | `FAILURE_TOLERANCE_EXCEEDED`   | `ALL_COMPLETED`                     |
     | `min_successful=N`               | `MIN_SUCCESSFUL_REACHED`       | `ALL_COMPLETED`                     |
 
-    !!! warning
+    The default `all_successful()` fails the batch on the first branch failure. Use
+    `CompletionConfig.all_completed()` to run every branch regardless of failures.
+
+    Set `should_complete` when the threshold fields cannot express the rule. The predicate
+    receives a `CompletionStatus` (`success_count`, `failure_count`, `completed_count`,
+    `total_count`, and `items`, a per-branch tuple of `CompletionItemStatus`) and returns a
+    `CompletionDecision`: `continue_batch()` to keep going, or `complete_batch(outcome)` to
+    stop, where `outcome` defaults to `CompletionOutcome.SUCCEEDED`. A
+    `CompletionOutcome.FAILED` outcome marks the whole batch failed, and `throw_if_error()`
+    then raises `BatchCompletionError` even when no individual branch failed.
+
+    ```python
+    --8<-- "examples/python/operations/parallel/custom-completion.py"
+    ```
 
-        `CompletionConfig.all_completed()` is deprecated. Use
-        `CompletionConfig.all_successful()` instead.
+    The predicate runs before any branch is scheduled (`completed_count == 0`) and again on
+    each terminal or suspension event, so it must handle the initial zero-progress
+    snapshot. It cannot be combined with `min_successful`, `tolerated_failure_count`, or
+    `tolerated_failure_percentage`, and must be deterministic, side-effect-free, and
+    monotonic. Unscheduled branches report `status=None` in `items`.
 
 === "Java"
 

examples/python/operations/map/custom-completion.py

@@ -0,0 +1,33 @@
+from aws_durable_execution_sdk_python import (
+    BatchResult,
+    CompletionStatus,
+    DurableContext,
+    complete_batch,
+    continue_batch,
+    durable_execution,
+)
+from aws_durable_execution_sdk_python.config import CompletionConfig, MapConfig
+
+
+def process_item(ctx: DurableContext, item: str, index: int, items: list[str]) -> str:
+    return ctx.step(lambda _: item.upper(), name=f"process-{index}")
+
+
+def stop_after_two_successes(status: CompletionStatus):
+    if status.success_count >= 2:
+        return complete_batch()
+    return continue_batch()
+
+
+@durable_execution
+def handler(event: dict, context: DurableContext) -> list[str]:
+    config = MapConfig(
+        completion_config=CompletionConfig(should_complete=stop_after_two_successes),
+    )
+    result: BatchResult[str] = context.map(
+        event["items"],
+        process_item,
+        name="process-items",
+        config=config,
+    )
+    return result.get_results()

examples/python/operations/parallel/custom-completion.py

@@ -0,0 +1,37 @@
+from aws_durable_execution_sdk_python import (
+    BatchResult,
+    CompletionStatus,
+    DurableContext,
+    complete_batch,
+    continue_batch,
+    durable_execution,
+)
+from aws_durable_execution_sdk_python.config import CompletionConfig, ParallelConfig
+
+
+def try_primary(ctx: DurableContext) -> str:
+    return ctx.step(lambda _: "primary result", name="primary")
+
+
+def try_secondary(ctx: DurableContext) -> str:
+    return ctx.step(lambda _: "secondary result", name="secondary")
+
+
+def stop_after_first_success(status: CompletionStatus):
+    if status.success_count >= 1:
+        return complete_batch()
+    return continue_batch()
+
+
+@durable_execution
+def handler(event: dict, context: DurableContext) -> str | None:
+    config = ParallelConfig(
+        completion_config=CompletionConfig(should_complete=stop_after_first_success),
+    )
+    result: BatchResult[str] = context.parallel(
+        [try_primary, try_secondary],
+        name="fetch-data",
+        config=config,
+    )
+    results = result.get_results()
+    return results[0] if results else None