feat: add distributed map operation

[nvasiu]

Summary

Adds the distributed map operation (ctx.distributed_map) to the Python SDK:
A map run processes a bounded dataset in parallel. A customer starts a map run from a durable function, naming a source to read items from, a processor function to invoke per batch, and concurrency, retry, and failure settings. The service reads items from the source, groups them into batches, invokes the processor for each batch, retries failures, tracks progress, routes successful results and failed items to destinations, and reports completion.

Changes

concurrency/models.py

• DistributedMapSummary: what ctx.distributed_map returns, describes run's overall outcome.
• DistributedMapResult: returned from ctx.distributed_map when inline result collection is enabled. Contains individual map run item outcomes.
• DistributedMapResultItem and DistributedMapItemError: represent a single item's result / error

config.py

• DistributedMapConfig: optional settings for distributed map
• DistributedMapSource: describes where map run items come from (inline list, S3, or a custom reader)
• DistributedMapProcessor: describes the Lambda that processes items and how outcomes are reported back
• ProcessorRetryConfig: configures how failing items are retried
• DistributedMapCompletionConfig: defines item failure thresholds for marking the overall map run failed
• SuccessDestination, FailureDestination, DistributedMapDestinationConfig, DistributedMapDestination: for routing successful and failed item records to S3

context.py

• ctx.distributed_map: the entry point a customer calls to run a distributed map

distributed_map_helpers.py

• Authoring wrappers: let a customer write a plain function and have it work as a processor Lambda without hand-writing the item or batch protocol, including durable-execution variants and a reader
• Currently placed in a separate top level file, can be moved elsewhere.

operation/distributed_map.py

• The executor: drives the operation so the caller's function suspends while the run executes and resumes with the finished outcome, and surfaces a clear error if the operation itself fails

lambda_service.py

• Carries the operation and its results to and from the backend service

state.py

• Stores the run's outcome in the durable execution state so it persists across suspend and resume

exceptions.py

• DistributedMapError: the error a customer catches when a run or an item fails

__init__.py

• Makes the distributed-map types importable by customers as public API

Tests

tests/operation/distributed_map_test.py
tests/context_test.py

• Core operation unit tests: executor, config/argument validation, wire round trips, result types

tests/e2e/distributed_map_int_test.py

• End to end ctx.distributed_map tests, mocking backend responses: suspend / resume, collect results, throw on failure

tests/distributed_map_helpers_test.py

• Authoring wrapper tests: checking that they process items, report failures, reject bad inputs

tests/e2e/distributed_map_helpers_int_test.py

• End to end authoring wrapper tests.

TODO

• When the model changes and distributed map implementation are complete in the durable service backend, these SDK changes need to be verified against them.

Future Tasks

• Add distributed map to the local emulator (in the testing package).
  • When this is done, we can add full end to end tests using the emulator.
• Add distributed map examples to the examples package.

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

[yaythomas]

The run status now lives on the operation. So a FAILED, STOPPED, or TIMED_OUT run arrives here with a details block. This branch raises before reading that block. So every non-SUCCEEDED run throws. _distributed_map_status_from_operation already maps these statuses, but this branch makes it unreachable. Pls resolve whenever details are present, and raise only when a terminal operation has none. The existing test only covers the no-details case.

[yaythomas]

This changes what every durable handler returns on a malformed or non-durable invoke. It returns a FAILED envelope where it used to raise. That is unrelated to distributed map and is not in the description. Pls split it into its own PR.

[yaythomas]

ruff format --check fails here and in tests/dmap_test.py. hatch fmt will fix it.

[yaythomas]

structure and contract

Although the functionality is mostly there, the structure doesn't integrate with the codebase's existing layering, so before a line-by-line review let's think through the broader organizational structure or shape of the code.

Why restructure

The codebase has two layers and one bridge between them:

1. config.py is the customer surface. Frozen dataclasses, validated on construction, no knowledge of the service API.
2. lambda_service.py is the service serialization layer. One frozen dataclass per API shape, named after the shape, each with from_dict / to_dict.
3. The executors in operation/ are the only code that touches both. InvokeOperationExecutor reads InvokeConfig and builds ChainedInvokeOptions. Nothing else translates.

The PR adds a third layer between the two: *Wire dataclasses in lambda_service.py, plus to_wire() methods on the config.py classes, plus a second set of translation functions (_source_to_wire, _build_distributed_map_options) in operation/dmap.py. Three consequences:

• Translation happens in two places, so a model change needs edits in both.
• config.py now imports sixteen serialization names at runtime. CONTRIBUTING.md names config as the lowest-level module. The dependency now points the wrong way.
• The *Wire suffix is a new naming convention. The existing API dataclasses carry the API shape's name (ChainedInvokeOptions, StepDetails, WaitOptions).

I sketch out an alternative layout below that removes the extra third layer. Each type gets one home and one job. Translation happens in exactly one place. The serialization layer mirrors the API model one to one, so a model change is a one-file edit. The customer types carry no serialization knowledge, so they validate and unit-test standalone. Envelope dataclasses replace the hand-built dicts in the handler wrappers, so the processor and reader protocols are typed and testable.

Proposed structure

config.py                     Customer surface. Frozen dataclasses. Validation in __post_init__ or the factory.
                              DistributedMapConfig, DistributedMapCompletionConfig, DistributedMapSource,
                              InlineSource / S3Source / ReaderSource (factories), DistributedMapProcessor,
                              SuccessDestination / FailureDestination, S3Destination (factories), S3Uri.
                              No to_wire(). Imports from lambda_service only the enums the customer sees
                              (CsvDelimiter, DistributedMapStatus, DistributedMapCompletionReason, DistributedMapItemStatus).

lambda_service.py             Serialization layer. One frozen dataclass per API shape, same name as the shape,
                              each with from_dict / to_dict: DistributedMapOptions, DistributedMapSourceConfig,
                              DistributedMapInlineSourceConfig, DistributedMapS3SourceConfig, DistributedMapCsvFormatOptions,
                              DistributedMapReaderFunctionSourceConfig, DistributedMapProcessorConfig,
                              DistributedMapDestinationConfig, DistributedMapOnSuccessConfig, DistributedMapOnFailureConfig,
                              DistributedMapS3DestinationConfig, DistributedMapCompletionConfig,
                              DistributedMapResultCollectionConfig, DistributedMapDetails, DistributedMapResultItem.
                              The enums. OperationUpdate.create_distributed_map_start(identifier, options).
                              Operation.distributed_map_details.

distributed_map/models.py     DistributedMapSummary, DistributedMapResult, DistributedMapResultItem, DistributedMapItemError.
                              Plain frozen dataclasses plus the accessors (succeeded(), failed(), get_results(),
                              get_errors(), throw_if_error()).

distributed_map/handlers.py   The create_* wrappers and the envelope dataclasses they read and write:
                              ProcessorEvent, ProcessorRecord, ItemHandlerResponse, ItemResult, ItemFailure,
                              ReaderEvent, ReaderResponse. Each with from_dict / to_dict.

operation/distributed_map.py  DistributedMapOperationExecutor. The only module that knows both layers.
                              Before the START checkpoint it translates the config types into DistributedMapOptions.
                              On resume it builds the summary or result from Operation.status plus
                              Operation.distributed_map_details. Same pattern as InvokeOperationExecutor.

context.py                    DurableContext.distributed_map, using _operation_replay_aware like invoke.
exceptions.py                 DistributedMapError.

Two names exist in both the customer surface and the API model: DistributedMapCompletionConfig and DistributedMapResultItem. Keep both. Alias the lambda_service one at the import site in the executor, as concurrency/models.py already does with BatchResult as BatchResultProtocol.

How the types flow:

 durable function                                       processor / reader Lambda
 ────────────────                                       ─────────────────────────
 ctx.distributed_map(source, processor,                 create_distributed_map_item_handler(fn)
                     max_concurrency, name, config)     create_distributed_map_reader(fn)
        │ config.py types                                       ▲
        ▼                                                       │ distributed_map/handlers.py
 operation/distributed_map.py                                   │ ProcessorEvent.from_dict(event)
   DistributedMapOperationExecutor                              │   → item_serdes.deserialize(record.body)
   │ translate config → lambda_service.DistributedMapOptions    │   → fn(item)
   │ OperationUpdate.create_distributed_map_start(...)          │   → ItemHandlerResponse.to_dict()
   │        └─ to_dict() ──► CheckpointDurableExecution ──► service reads the source, invokes ──┘
   │                                                            the processor per batch, writes destinations
   │ suspend
   │
   │ resume: Operation.from_dict ──► Operation.status, Operation.distributed_map_details
   │         result_serdes.deserialize(item.output) for each result item
   ▼
 distributed_map/models.py  DistributedMapSummary or DistributedMapResult ──► returned to the caller

The translation the executor owns:

Customer surface (config.py)	API model (lambda_service.py)
plain list, InlineSource.of(items, serdes=, max_items=)	DistributedMapSourceConfig(type=INLINE, inline_source_config=Items[str]); each item serialized with the item serdes
S3Source.json_lines / json_array / csv(uri)	S3SourceConfig(key=…, format=…); headers given → CsvFormatOptions(header_location=GIVEN), else FIRST_ROW
S3Source.objects / flattened_*(prefix_uri)	S3SourceConfig(key_prefix=…, transform=NONE / LOAD_AND_FLATTEN)
ReaderSource.from_function(name, initial_state=, state_serdes=)	ReaderFunctionSourceConfig(function_name, initial_state=state_serdes.serialize(...))
DistributedMapProcessor.batch / item_failures / item_results	FunctionResponseTypes omitted / [REPORT_BATCH_ITEM_FAILURES] / [REPORT_BATCH_ITEM_RESULTS]
max_retry_attempts="unlimited"	MaxRetryAttempts=-1
max_retry_duration: Duration	MaxRetryDurationSeconds
S3Destination.successes(include_input, include_output)	OnSuccessConfig(type=S3, include=[INPUT?, OUTPUT?]); failures likewise with ERROR
.failure_count(n) / .failure_percentage(p, minimum_sample_size)	CompletionConfig(ToleratedFailureCount) / (ToleratedFailurePercentage, MinimumSampleSize)
DistributedMapResultConfig (see below)	ResultCollectionConfig(mode=INLINE)
timeout: Duration	TimeoutSeconds

Return type. DistributedMapResult extends DistributedMapSummary, so declare distributed_map as returning DistributedMapSummary, and make result collection static with @overload: a DistributedMapResultConfig(DistributedMapConfig) that adds result_serdes selects the DistributedMapResult overload, and a plain DistributedMapConfig selects the summary. Callers then never need isinstance.

Handler wrappers as decorators

Make the wrappers decorators, named the way this repo names its decorators (durable_execution, durable_step): drop create_, and mark the durable variants with the durable_ prefix instead of the _with_durable_execution suffix.

PR name	Proposed
create_distributed_map_item_handler	distributed_map_item_handler
create_distributed_map_batch_handler	distributed_map_batch_handler
create_distributed_map_reader	distributed_map_reader
create_distributed_map_item_handler_with_durable_execution	durable_distributed_map_item_handler
create_distributed_map_batch_handler_with_durable_execution	durable_distributed_map_batch_handler

durable_ means what it means on durable_execution: the decorated function is a durable function and the Lambda must be deployed as one. Give them the same signature shape as durable_execution (func optional, options keyword-only) so all three forms work:

@durable_distributed_map_item_handler
def handler(ctx: DurableContext, order: dict) -> dict: ...

@distributed_map_item_handler(report="failures", concurrency=8)
def handler(order: dict) -> None: ...

handler = distributed_map_reader(read_page, state_serdes=my_serdes)

The decorated name becomes the Lambda handler and takes (event, context), so the natural name for the decorated function is handler. Say so in the docstring.

Other issues with PR currently

Contract:

1. A failed run raises. check_result_status calls raise_operation_error(DistributedMapError) when the operation is FAILED, TIMED_OUT, or STOPPED. The DistributedMapSummary docstring in this PR states the intended contract: a non-SUCCEEDED run resolves with the summary, and throw_if_error() opts in to raising. The executor must build and return the summary for those statuses. The e2e tests do not catch this because they inject Operation.Status=SUCCEEDED together with a DistributedMapDetails.Status field that the API model does not define. Status lives on the operation.
2. Items, Output, and the record body are handled as JSON values. _inline_items_to_wire does json.loads(serialized) and rejects a serdes whose output is not JSON. _deserialize_items does json.dumps(wire.output) before deserializing. _to_item in the wrappers does json.dumps(body) before deserializing. The API model types Items entries and Output as strings. Pass the serdes output through unchanged, and hand the string to the serdes unchanged on the way back. This also restores support for non-JSON serdes.
3. distributed_map_id is arn.rsplit(":", 1)[-1]. The run id is the segment after the last /distributed-map-run/. The current code returns $LATEST/durable-execution/….
4. Factory names differ from the names agreed across the SDKs: report_batch_outcome / report_failed_items / report_item_results should be batch / item_failures / item_results; DistributedMapSource.S3.json_lines should be S3Source.json_lines; DistributedMapDestination.S3.successes should be S3Destination.successes; DistributedMapSource.inline should be InlineSource.of; ProcessorRetryConfig should be max_retry_attempts / max_retry_duration options on the processor factory; ReaderSource.from_function is missing.
5. Eighteen names are exported from the package __init__. The package root is the common import path for every durable function. Keep this feature's factories, wrappers, and result types in distributed_map, and export nothing new from the root.
6. The return type is a union, DistributedMapSummary | DistributedMapResult, which forces isinstance at every call site. See the overload proposal above.

Style (CONTRIBUTING.md):

• Translation in two places (to_wire() on config classes, and _source_to_wire / _build_distributed_map_options in the executor). Move it all to the executor.
• config.py imports sixteen serialization names at runtime. After the move it needs only the customer-visible enums.
• *Wire suffix. Name the serialization dataclasses after the API shapes, like the existing ones.
• _summary_fields() returns an ad hoc dict that is splatted into a constructor. _error_entry, _item_response, and the wrapper bodies build response dicts by hand. Use envelope dataclasses with to_dict.
• dmap.py holds a module global _CTX = SerDesContext(). Build the context per call.
• dmap as a module name. The codebase spells names out (wait_for_condition, concurrency), and the method is distributed_map.
• The branch is behind main. It calls _replay_aware() and _create_step_id(); main has _operation_replay_aware(sub_type, name).
• The PR diff carries 159 files that are not part of this feature, from the fork's stale main. Only 19 files are the feature. The execution.py change is unrelated to it. Please rebase onto upstream main so the diff is the feature alone.