[WIP] Add experimental Event Groups support

CHANGELOG.md

@@ -20,6 +20,13 @@ to include examples, links to docs, or any other relevant information.
 
 ### Added
 
+- **Experimental**: Event Groups tag logically related commands so that the UI and CLI can
+  visualize, analyze, and debug them together. Create a group with
+  `workflow.create_event_group(label)`, then attach it either per call
+  (`workflow.start_activity(..., event_groups=[group])`) or ambiently to everything issued inside
+  `with group.scope():`. Each signal and update handler is also implicitly wrapped in a group of
+  its own. Requires a server that understands the Event Groups fields.
+
 ### Changed
 
 - System Nexus Signal-with-Start Workflow operations now use the typed

README.md

@@ -976,6 +976,49 @@ await workflow.wait_condition(workflow.all_handlers_finished)
 * `await handle.signal()` can be called on the handle to signal the external workflow
 * `await handle.cancel()` can be called on the handle to send a cancel to the external workflow
 
+#### Event Groups
+
+Event Groups regroup logically related events of a Workflow Execution's history, so that UIs and other tools can
+present them together. A group is created with `workflow.create_event_group(label)` and can be attached to the
+commands a workflow produces, either explicitly through the `event_groups` option of the API producing the command,
+or implicitly to every command produced within `group.scope()`:
+
+```python
+@workflow.defn
+class MyWorkflow:
+    @workflow.run
+    async def run(self) -> None:
+        payment_group = workflow.create_event_group("payment-processing")
+        customer_group = workflow.create_event_group(
+            "customer-james-watkins", id="customer-123456"
+        )
+
+        # Explicit attachment of Event Groups to a single command
+        await workflow.execute_activity(
+            my_activity,
+            arg,
+            start_to_close_timeout=timedelta(minutes=1),
+            event_groups=[payment_group, customer_group],
+        )
+
+        # Scope-based propagation, applying to every command produced in the block
+        with payment_group.scope(), customer_group.scope():
+            await authorize_payment(...)
+            await capture_payment(...)
+```
+
+Scopes nest, and coroutines started inside a scope inherit it, since they capture the context active at their
+creation. Two Event Groups group events together if and only if they have the same id; by default the id is derived
+deterministically from the label, so two groups created with the same label in the same execution are the same group.
+Pass an explicit `id` to distinguish groups that share a label, or to group events under a business identifier. Note
+that a derived id is a hash of the label, so avoid putting sensitive information in labels of groups without an
+explicit id.
+
+The SDK also creates Event Groups implicitly around the workflow main method, signal handlers, and update handlers, so
+that the commands they produce are grouped with the event that triggered them.
+
+WARNING: Event Groups is an experimental API and may change without notice.
+
 #### Testing
 
 Workflow testing can be done in an integration-test fashion against a real server, however it is hard to simulate

temporalio/worker/_interceptor.py

@@ -177,6 +177,7 @@ class ContinueAsNewInput:
     headers: Mapping[str, temporalio.api.common.v1.Payload]
     versioning_intent: VersioningIntent | None
     initial_versioning_behavior: ContinueAsNewVersioningBehavior | None
+    event_groups: Sequence[temporalio.workflow.EventGroup] | None
     # The types may be absent
     arg_types: list[type] | None
 
@@ -263,6 +264,7 @@ class StartActivityInput:
     disable_eager_execution: bool
     versioning_intent: VersioningIntent | None
     summary: str | None
+    event_groups: Sequence[temporalio.workflow.EventGroup] | None
     priority: temporalio.common.Priority
     # The types may be absent
     arg_types: list[type] | None
@@ -293,6 +295,7 @@ class StartChildWorkflowInput:
     versioning_intent: VersioningIntent | None
     static_summary: str | None
     static_details: str | None
+    event_groups: Sequence[temporalio.workflow.EventGroup] | None
     priority: temporalio.common.Priority
     # The types may be absent
     arg_types: list[type] | None
@@ -313,6 +316,7 @@ class StartNexusOperationInput(Generic[InputT, OutputT]):
     cancellation_type: temporalio.workflow.NexusOperationCancellationType
     headers: Mapping[str, str] | None
     summary: str | None
+    event_groups: Sequence[temporalio.workflow.EventGroup] | None = None
     output_type: type[OutputT] | None = None
 
     def __post_init__(self) -> None:
@@ -366,6 +370,7 @@ class StartLocalActivityInput:
     cancellation_type: temporalio.workflow.ActivityCancellationType
     headers: Mapping[str, temporalio.api.common.v1.Payload]
     summary: str | None
+    event_groups: Sequence[temporalio.workflow.EventGroup] | None
 
     # The types may be absent
     arg_types: list[type] | None

temporalio/worker/_workflow.py

@@ -694,6 +694,7 @@ def _create_workflow_instance(
             first_execution_run_id=init.first_execution_run_id,
             headers=dict(init.headers),
             namespace=self._namespace,
+            original_execution_run_id=init.original_execution_run_id or act.run_id,
             parent=parent,
             root=root,
             raw_memo=dict(init.memo.fields),