From 1b1a3e533781db8a1262119d16d079513eae8e1b Mon Sep 17 00:00:00 2001
From: James Watkins-Harvey <james.watkinsharvey@temporal.io>
Date: Mon, 17 Aug 2026 15:33:34 -0400
Subject: [PATCH] Add experimental Event Groups support

---
 CHANGELOG.md                                  |    7 +
 README.md                                     |   43 +
 temporalio/worker/_interceptor.py             |    5 +
 temporalio/worker/_workflow.py                |    1 +
 temporalio/worker/_workflow_instance.py       |   99 +-
 temporalio/worker/workflow_sandbox/_runner.py |    1 +
 temporalio/workflow/__init__.py               |   12 +
 temporalio/workflow/_activities.py            |  103 +
 temporalio/workflow/_context.py               |   37 +-
 temporalio/workflow/_event_groups.py          |  242 +++
 temporalio/workflow/_nexus.py                 |   23 +-
 temporalio/workflow/_workflow_ops.py          |   21 +
 tests/test_workflow_exports.py                |    5 +
 tests/worker/test_event_groups.py             | 1932 +++++++++++++++++
 14 files changed, 2516 insertions(+), 15 deletions(-)
 create mode 100644 temporalio/workflow/_event_groups.py
 create mode 100644 tests/worker/test_event_groups.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index d0d80bc71..f2bdb27c4 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/README.md b/README.md
index 7a1caedd9..b7b1810c7 100644
--- a/README.md
+++ b/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
diff --git a/temporalio/worker/_interceptor.py b/temporalio/worker/_interceptor.py
index f59b534c2..046a58937 100644
--- a/temporalio/worker/_interceptor.py
+++ b/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
diff --git a/temporalio/worker/_workflow.py b/temporalio/worker/_workflow.py
index 1b217b4a5..f50955c4c 100644
--- a/temporalio/worker/_workflow.py
+++ b/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),
diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py
index fc5a6efe3..dddd4315f 100644
--- a/temporalio/worker/_workflow_instance.py
+++ b/temporalio/worker/_workflow_instance.py
@@ -722,7 +722,8 @@ async def run_update() -> None:
                 command = None  # type: ignore
 
                 # Run the handler
-                success = await self._inbound.handle_update_handler(handler_input)
+                with temporalio.workflow._inbound_update_event_group(job.id).scope():
+                    success = await self._inbound.handle_update_handler(handler_input)
                 result_payloads = self._workflow_context_payload_converter.to_payloads(
                     [success]
                 )
@@ -1144,7 +1145,7 @@ def _apply_signal_workflow(
         self._process_signal_job(signal_defn, job)
 
     def _apply_initialize_workflow(
-        self, _job: temporalio.bridge.proto.workflow_activation.InitializeWorkflow
+        self, job: temporalio.bridge.proto.workflow_activation.InitializeWorkflow
     ) -> None:
         # Async call to run on the scheduler thread. This will be wrapped in
         # another function which applies exception handling.
@@ -1241,6 +1242,7 @@ def workflow_continue_as_new(
         versioning_intent: temporalio.workflow.VersioningIntent | None,
         initial_versioning_behavior: temporalio.workflow.ContinueAsNewVersioningBehavior
         | None,
+        event_groups: Sequence[temporalio.workflow.EventGroup] | None = None,
     ) -> NoReturn:
         self._assert_not_read_only("continue as new")
         # Use definition if callable
@@ -1270,6 +1272,7 @@ def workflow_continue_as_new(
                 arg_types=arg_types,
                 versioning_intent=versioning_intent,
                 initial_versioning_behavior=initial_versioning_behavior,
+                event_groups=event_groups,
             )
         )
 
@@ -1395,6 +1398,7 @@ def workflow_upsert_memo(self, updates: Mapping[str, Any]) -> None:
             return
 
         command = self._add_command()
+        command.event_group_markers.extend(self._event_group_markers(None))
         fields = command.modify_workflow_properties.upserted_memo.fields
 
         # Updating memo inside info by downcasting to mutable mapping.
@@ -1463,6 +1467,7 @@ def workflow_patch(self, id: str, *, deprecated: bool) -> bool:
         self._patches_memoized[id] = use_patch
         if use_patch:
             command = self._add_command()
+            command.event_group_markers.extend(self._event_group_markers(None))
             command.set_patch_marker.patch_id = id
             command.set_patch_marker.deprecated = deprecated
         return use_patch
@@ -1561,6 +1566,7 @@ def workflow_start_activity(
         activity_id: str | None,
         versioning_intent: temporalio.workflow.VersioningIntent | None,
         summary: str | None = None,
+        event_groups: Sequence[temporalio.workflow.EventGroup] | None = None,
         priority: temporalio.common.Priority = temporalio.common.Priority.default,
     ) -> temporalio.workflow.ActivityHandle[Any]:
         self._assert_not_read_only("start activity")
@@ -1598,6 +1604,7 @@ def workflow_start_activity(
                 ret_type=ret_type,
                 versioning_intent=versioning_intent,
                 summary=summary,
+                event_groups=event_groups,
                 priority=priority,
             )
         )
@@ -1626,6 +1633,7 @@ async def workflow_start_child_workflow(
         versioning_intent: temporalio.workflow.VersioningIntent | None,
         static_summary: str | None = None,
         static_details: str | None = None,
+        event_groups: Sequence[temporalio.workflow.EventGroup] | None = None,
         priority: temporalio.common.Priority = temporalio.common.Priority.default,
     ) -> temporalio.workflow.ChildWorkflowHandle[Any, Any]:
         # Use definition if callable
@@ -1666,6 +1674,7 @@ async def workflow_start_child_workflow(
                 versioning_intent=versioning_intent,
                 static_summary=static_summary,
                 static_details=static_details,
+                event_groups=event_groups,
                 priority=priority,
             )
         )
@@ -1683,6 +1692,7 @@ def workflow_start_local_activity(
         cancellation_type: temporalio.workflow.ActivityCancellationType,
         activity_id: str | None,
         summary: str | None,
+        event_groups: Sequence[temporalio.workflow.EventGroup] | None = None,
     ) -> temporalio.workflow.ActivityHandle[Any]:
         # Get activity definition if it's callable
         name: str
@@ -1716,6 +1726,7 @@ def workflow_start_local_activity(
                 local_retry_threshold=local_retry_threshold,
                 cancellation_type=cancellation_type,
                 summary=summary,
+                event_groups=event_groups,
                 headers={},
                 arg_types=arg_types,
                 ret_type=ret_type,
@@ -1735,6 +1746,7 @@ async def workflow_start_nexus_operation(
         cancellation_type: temporalio.workflow.NexusOperationCancellationType,
         headers: Mapping[str, str] | None,
         summary: str | None,
+        event_groups: Sequence[temporalio.workflow.EventGroup] | None = None,
     ) -> temporalio.workflow.NexusOperationHandle[OutputT]:
         if temporalio.nexus.system.is_system_endpoint(endpoint):
             return await _start_system_nexus_operation(
@@ -1766,6 +1778,7 @@ async def workflow_start_nexus_operation(
                 cancellation_type=cancellation_type,
                 headers=headers,
                 summary=summary,
+                event_groups=event_groups,
             )
         )
 
@@ -1779,7 +1792,9 @@ def workflow_upsert_search_attributes(
             | Sequence[temporalio.common.SearchAttributeUpdate]
         ),
     ) -> None:
-        v = self._add_command().upsert_workflow_search_attributes
+        command = self._add_command()
+        command.event_group_markers.extend(self._event_group_markers(None))
+        v = command.upsert_workflow_search_attributes
 
         # Update the attrs on info, casting to their mutable forms first
         mut_attrs = cast(
@@ -1879,7 +1894,11 @@ def workflow_upsert_search_attributes(
                     )
 
     async def workflow_sleep(
-        self, duration: float, *, summary: str | None = None
+        self,
+        duration: float,
+        *,
+        summary: str | None = None,
+        event_groups: Sequence[temporalio.workflow.EventGroup] | None = None,
     ) -> None:
         user_metadata = (
             temporalio.api.sdk.v1.UserMetadata(
@@ -1891,7 +1910,7 @@ async def workflow_sleep(
         fut = self.create_future()
         timer_handle = self._timer_impl(
             duration,
-            _TimerOptions(user_metadata=user_metadata),
+            _TimerOptions(user_metadata=user_metadata, event_groups=event_groups),
             lambda: fut.set_result(None) if not fut.done() else None,
         )
         fut.add_done_callback(
@@ -1905,6 +1924,7 @@ async def workflow_wait_condition(
         *,
         timeout: float | None = None,
         timeout_summary: str | None = None,
+        event_groups: Sequence[temporalio.workflow.EventGroup] | None = None,
     ) -> None:
         self._assert_not_read_only("wait condition")
         cancellation_requested_before = self._cancel_reason is not None
@@ -1934,7 +1954,9 @@ def cancellation_arrived() -> bool:
         ctxvars = contextvars.copy_context()
 
         async def in_context():
-            _TimerOptionsCtxVar.set(_TimerOptions(user_metadata=user_metadata))
+            _TimerOptionsCtxVar.set(
+                _TimerOptions(user_metadata=user_metadata, event_groups=event_groups)
+            )
             await asyncio.wait_for(fut, timeout)
 
         try:
@@ -2039,10 +2061,12 @@ async def run_activity() -> Any:
                         completed_cancellation_flag=_WorkflowLogicFlag.RAISE_ON_CANCELLING_COMPLETED_ACTIVITY,
                     )
                 except _ActivityDoBackoffError as err:
-                    # We have to sleep then reschedule. Note this sleep can be
-                    # cancelled like any other timer.
-                    await asyncio.sleep(
-                        err.backoff.backoff_duration.ToTimedelta().total_seconds()
+                    # Use workflow_sleep rather than asyncio.sleep so directly
+                    # attached Event Groups on the local activity are copied onto
+                    # the backoff timer, matching the command being retried.
+                    await self.workflow_sleep(
+                        err.backoff.backoff_duration.ToTimedelta().total_seconds(),
+                        event_groups=input.event_groups,
                     )
                     handle._apply_schedule_command(err.backoff)
                     # We have to put the handle back on the pending activity
@@ -2066,6 +2090,7 @@ async def _outbound_signal_child_workflow(
         )
         payloads = payload_converter.to_payloads(input.args) if input.args else None
         command = self._add_command()
+        command.event_group_markers.extend(self._event_group_markers(None))
         v = command.signal_external_workflow_execution
         v.child_workflow_id = input.child_workflow_id
         v.signal_name = input.signal
@@ -2086,6 +2111,7 @@ async def _outbound_signal_external_workflow(
         )
         payloads = payload_converter.to_payloads(input.args) if input.args else None
         command = self._add_command()
+        command.event_group_markers.extend(self._event_group_markers(None))
         v = command.signal_external_workflow_execution
         v.workflow_execution.namespace = input.namespace
         v.workflow_execution.workflow_id = input.workflow_id
@@ -2216,6 +2242,16 @@ def _add_command(self) -> temporalio.bridge.proto.workflow_commands.WorkflowComm
         self._assert_not_read_only("add command")
         return self._current_completion.successful.commands.add()
 
+    def _event_group_markers(
+        self, event_groups: Sequence[temporalio.workflow.EventGroup] | None
+    ) -> list[temporalio.api.sdk.v1.EventGroupMarker]:
+        """Snapshot the Event Groups for a command being requested.
+
+        Must be called while still in the requesting code's context, which may
+        differ from the one the command is built in.
+        """
+        return temporalio.workflow._event_group_markers_to_proto(event_groups)
+
     def _workflow_logic_flag_enabled(self, flag: _WorkflowLogicFlag) -> bool:
         if flag in self._current_internal_flags:
             return True
@@ -2655,8 +2691,12 @@ def _process_signal_job(
         def done_callback(_f: Any):
             self._in_progress_signals.pop(id, None)
 
+        async def run_signal() -> None:
+            with _implicit_event_group_scope(job.originating_event_id):
+                await self._inbound.handle_signal(input)
+
         task = self.create_task(
-            self._run_top_level_workflow_function(self._inbound.handle_signal(input)),
+            self._run_top_level_workflow_function(run_signal()),
             name=f"signal: {job.signal_name}",
         )
         task.add_done_callback(done_callback)
@@ -2928,7 +2968,14 @@ def _timer_impl(
         # Create, schedule, and return
         seq = self._next_seq("timer")
         handle = _TimerHandle(
-            seq, self.time() + delay, options, callback, args, self, context
+            seq,
+            self.time() + delay,
+            options,
+            callback,
+            args,
+            self,
+            context,
+            self._event_group_markers(options.event_groups if options else None),
         )
         handle._apply_start_command(self._add_command(), delay)
         self._pending_timers[seq] = handle
@@ -3218,9 +3265,24 @@ def start_local_activity(
         return self._instance._outbound_schedule_activity(input)
 
 
+@contextmanager
+def _implicit_event_group_scope(originating_event_id: int) -> Iterator[None]:
+    """Enter the scope of the implicit Event Group of an inbound signal.
+
+    No group is created if the activation does not identify the originating
+    event, which happens with servers predating Event Groups.
+    """
+    if not originating_event_id:
+        yield
+        return
+    with temporalio.workflow._inbound_event_group(originating_event_id).scope():
+        yield
+
+
 @dataclass(frozen=True)
 class _TimerOptions:
     user_metadata: temporalio.api.sdk.v1.UserMetadata | None = None
+    event_groups: Sequence[temporalio.workflow.EventGroup] | None = None
 
 
 _TimerOptionsCtxVar: contextvars.ContextVar[_TimerOptions] = contextvars.ContextVar(
@@ -3238,10 +3300,12 @@ def __init__(
         args: Sequence[Any],
         loop: asyncio.AbstractEventLoop,
         context: contextvars.Context | None,
+        event_group_markers: Sequence[temporalio.api.sdk.v1.EventGroupMarker] = (),
     ) -> None:
         super().__init__(when, callback, args, loop, context)
         self._seq = seq
         self._options = options
+        self._event_group_markers = event_group_markers
 
     def _apply_start_command(
         self,
@@ -3251,6 +3315,7 @@ def _apply_start_command(
         command.start_timer.seq = self._seq
         if self._options and self._options.user_metadata:
             command.user_metadata.CopyFrom(self._options.user_metadata)
+        command.event_group_markers.extend(self._event_group_markers)
         command.start_timer.start_to_fire_timeout.FromNanoseconds(int(delay * 1e9))
 
     def _apply_cancel_command(
@@ -3258,6 +3323,7 @@ def _apply_cancel_command(
         command: temporalio.bridge.proto.workflow_commands.WorkflowCommand,
     ) -> None:
         command.cancel_timer.seq = self._seq
+        command.event_group_markers.extend(self._event_group_markers)
 
 
 class _ActivityDoBackoffError(BaseException):
@@ -3297,6 +3363,7 @@ def __init__(
                 is_local=isinstance(self._input, StartLocalActivityInput),
             )
         )
+        self._event_group_markers = instance._event_group_markers(input.event_groups)
 
     def cancel(self, msg: Any | None = None) -> bool:
         # Allow the cancel to go through for the task even if we're deleting,
@@ -3355,6 +3422,7 @@ def _apply_schedule_command(
             if isinstance(self._input, StartLocalActivityInput)
             else command.schedule_activity
         )
+        command.event_group_markers.extend(self._event_group_markers)
         v.seq = self._seq
         v.activity_id = self._input.activity_id or str(self._seq)
         v.activity_type = self._input.activity
@@ -3450,6 +3518,7 @@ def __init__(
         self._failure_converter = self._instance._failure_converter_with_context(
             workflow_context
         )
+        self._event_group_markers = instance._event_group_markers(input.event_groups)
 
     @property
     def id(self) -> str:
@@ -3506,6 +3575,7 @@ def _apply_start_command(self) -> None:
         )
 
         command = self._instance._add_command()
+        command.event_group_markers.extend(self._event_group_markers)
         v = command.start_child_workflow_execution
         v.seq = self._seq
         v.namespace = self._instance._info.namespace
@@ -3613,6 +3683,7 @@ async def signal(
     async def cancel(self, *, reason: str = "") -> None:
         self._instance._assert_not_read_only("cancel external handle")
         command = self._instance._add_command()
+        command.event_group_markers.extend(self._instance._event_group_markers(None))
         v = command.request_cancel_external_workflow_execution
         v.workflow_execution.namespace = self._instance._info.namespace
         v.workflow_execution.workflow_id = self._id
@@ -3639,6 +3710,7 @@ def __init__(
         self._result_fut: asyncio.Future[OutputT | None] = instance.create_future()
         self._payload_converter = payload_converter
         self._failure_converter = self._instance._context_free_failure_converter
+        self._event_group_markers = instance._event_group_markers(input.event_groups)
 
     @property
     def operation_token(self) -> str | None:
@@ -3673,6 +3745,7 @@ def _resolve_failure(self, err: BaseException) -> None:
     def _apply_schedule_command(self) -> None:
         payload = self._payload_converter.to_payload(self._input.input)
         command = self._instance._add_command()
+        command.event_group_markers.extend(self._event_group_markers)
         v = command.schedule_nexus_operation
         v.seq = self._seq
         v.endpoint = self._input.endpoint
@@ -3717,6 +3790,7 @@ def __init__(
         super().__init__("Continue as new")
         self._instance = instance
         self._input = input
+        self._event_group_markers = instance._event_group_markers(input.event_groups)
 
     def _apply_command(self) -> None:
         # Convert arguments before creating command in case it raises error
@@ -3739,6 +3813,7 @@ def _apply_command(self) -> None:
         )
 
         command = self._instance._add_command()
+        command.event_group_markers.extend(self._event_group_markers)
         v = command.continue_as_new_workflow_execution
         v.SetInParent()
         if self._input.workflow:
diff --git a/temporalio/worker/workflow_sandbox/_runner.py b/temporalio/worker/workflow_sandbox/_runner.py
index 7f06bfcd6..89ce7b951 100644
--- a/temporalio/worker/workflow_sandbox/_runner.py
+++ b/temporalio/worker/workflow_sandbox/_runner.py
@@ -41,6 +41,7 @@
     first_execution_run_id="sandbox-validate-first-run_id",
     headers={},
     namespace="sandbox-validate-namespace",
+    original_execution_run_id="sandbox-validate-original-execution-run_id",
     parent=None,
     root=None,
     raw_memo={},
diff --git a/temporalio/workflow/__init__.py b/temporalio/workflow/__init__.py
index fa2681139..e26d5af05 100644
--- a/temporalio/workflow/__init__.py
+++ b/temporalio/workflow/__init__.py
@@ -107,6 +107,13 @@
     init,
     run,
 )
+from ._event_groups import (
+    EventGroup,
+    _event_group_markers_to_proto,
+    _inbound_event_group,
+    _inbound_update_event_group,
+    create_event_group,
+)
 from ._exceptions import (
     ContinueAsNewVersioningBehavior,
     NondeterminismError,
@@ -228,6 +235,8 @@
     "uuid4",
     "uuid7",
     "wait_condition",
+    "EventGroup",
+    "create_event_group",
     "DynamicWorkflowConfig",
     "defn",
     "dynamic_config",
@@ -281,6 +290,9 @@
     "_release_waiter",
     "_wait",
     "_current_update_info",
+    "_event_group_markers_to_proto",
+    "_inbound_event_group",
+    "_inbound_update_event_group",
     "_Runtime",
     "_set_current_update_info",
     "_Definition",
diff --git a/temporalio/workflow/_activities.py b/temporalio/workflow/_activities.py
index ef883c016..36260e45e 100644
--- a/temporalio/workflow/_activities.py
+++ b/temporalio/workflow/_activities.py
@@ -25,6 +25,7 @@
     SelfType,
 )
 from ._context import _Runtime
+from ._event_groups import EventGroup
 from ._exceptions import VersioningIntent
 
 __all__ = [
@@ -97,6 +98,7 @@ class ActivityConfig(TypedDict, total=False):
     activity_id: str | None
     versioning_intent: VersioningIntent | None
     summary: str | None
+    event_groups: Sequence[EventGroup] | None
     priority: temporalio.common.Priority
 
 
@@ -115,6 +117,7 @@ def start_activity(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ActivityHandle[ReturnType]: ...
 
@@ -134,6 +137,7 @@ def start_activity(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ActivityHandle[ReturnType]: ...
 
@@ -154,6 +158,7 @@ def start_activity(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ActivityHandle[ReturnType]: ...
 
@@ -174,6 +179,7 @@ def start_activity(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ActivityHandle[ReturnType]: ...
 
@@ -194,6 +200,7 @@ def start_activity(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ActivityHandle[ReturnType]: ...
 
@@ -214,6 +221,7 @@ def start_activity(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ActivityHandle[ReturnType]: ...
 
@@ -236,6 +244,7 @@ def start_activity(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ActivityHandle[Any]: ...
 
@@ -256,6 +265,7 @@ def start_activity(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ActivityHandle[Any]:
     """Start an activity and return its handle.
@@ -293,6 +303,9 @@ def start_activity(
             Deprecated: Use Worker Deployment versioning instead.
         summary: A single-line fixed summary for this activity that may appear in UI/CLI.
             This can be in single-line Temporal markdown format.
+        event_groups: Event Groups to associate this command with, in
+            addition to those active in the current scope. See
+            :py:func:`temporalio.workflow.create_event_group`.
         priority: Priority of the activity.
 
     Returns:
@@ -312,6 +325,7 @@ def start_activity(
         activity_id=activity_id,
         versioning_intent=versioning_intent,
         summary=summary,
+        event_groups=event_groups,
         priority=priority,
     )
 
@@ -331,6 +345,7 @@ async def execute_activity(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ReturnType: ...
 
@@ -350,6 +365,7 @@ async def execute_activity(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ReturnType: ...
 
@@ -370,6 +386,7 @@ async def execute_activity(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ReturnType: ...
 
@@ -390,6 +407,7 @@ async def execute_activity(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ReturnType: ...
 
@@ -410,6 +428,7 @@ async def execute_activity(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ReturnType: ...
 
@@ -430,6 +449,7 @@ async def execute_activity(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ReturnType: ...
 
@@ -452,6 +472,7 @@ async def execute_activity(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> Any: ...
 
@@ -472,6 +493,7 @@ async def execute_activity(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> Any:
     """Start an activity and wait for completion.
@@ -494,6 +516,7 @@ async def execute_activity(
         activity_id=activity_id,
         versioning_intent=versioning_intent,
         summary=summary,
+        event_groups=event_groups,
         priority=priority,
     )
 
@@ -513,6 +536,7 @@ def start_activity_class(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ActivityHandle[ReturnType]: ...
 
@@ -532,6 +556,7 @@ def start_activity_class(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ActivityHandle[ReturnType]: ...
 
@@ -552,6 +577,7 @@ def start_activity_class(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ActivityHandle[ReturnType]: ...
 
@@ -572,6 +598,7 @@ def start_activity_class(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ActivityHandle[ReturnType]: ...
 
@@ -592,6 +619,7 @@ def start_activity_class(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ActivityHandle[ReturnType]: ...
 
@@ -612,6 +640,7 @@ def start_activity_class(  # type: ignore[reportOverlappingOverload]
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ActivityHandle[ReturnType]: ...
 
@@ -631,6 +660,7 @@ def start_activity_class(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ActivityHandle[Any]:
     """Start an activity from a callable class.
@@ -651,6 +681,7 @@ def start_activity_class(
         activity_id=activity_id,
         versioning_intent=versioning_intent,
         summary=summary,
+        event_groups=event_groups,
         priority=priority,
     )
 
@@ -670,6 +701,7 @@ async def execute_activity_class(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ReturnType: ...
 
@@ -689,6 +721,7 @@ async def execute_activity_class(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ReturnType: ...
 
@@ -709,6 +742,7 @@ async def execute_activity_class(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ReturnType: ...
 
@@ -729,6 +763,7 @@ async def execute_activity_class(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ReturnType: ...
 
@@ -749,6 +784,7 @@ async def execute_activity_class(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ReturnType: ...
 
@@ -769,6 +805,7 @@ async def execute_activity_class(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ReturnType: ...
 
@@ -788,6 +825,7 @@ async def execute_activity_class(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> Any:
     """Start an activity from a callable class and wait for completion.
@@ -808,6 +846,7 @@ async def execute_activity_class(
         activity_id=activity_id,
         versioning_intent=versioning_intent,
         summary=summary,
+        event_groups=event_groups,
         priority=priority,
     )
 
@@ -827,6 +866,7 @@ def start_activity_method(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ActivityHandle[ReturnType]: ...
 
@@ -846,6 +886,7 @@ def start_activity_method(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ActivityHandle[ReturnType]: ...
 
@@ -866,6 +907,7 @@ def start_activity_method(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ActivityHandle[ReturnType]: ...
 
@@ -886,6 +928,7 @@ def start_activity_method(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ActivityHandle[ReturnType]: ...
 
@@ -906,6 +949,7 @@ def start_activity_method(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ActivityHandle[ReturnType]: ...
 
@@ -926,6 +970,7 @@ def start_activity_method(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ActivityHandle[ReturnType]: ...
 
@@ -945,6 +990,7 @@ def start_activity_method(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ActivityHandle[Any]:
     """Start an activity from a method.
@@ -965,6 +1011,7 @@ def start_activity_method(
         activity_id=activity_id,
         versioning_intent=versioning_intent,
         summary=summary,
+        event_groups=event_groups,
         priority=priority,
     )
 
@@ -984,6 +1031,7 @@ async def execute_activity_method(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ReturnType: ...
 
@@ -1003,6 +1051,7 @@ async def execute_activity_method(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ReturnType: ...
 
@@ -1023,6 +1072,7 @@ async def execute_activity_method(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ReturnType: ...
 
@@ -1043,6 +1093,7 @@ async def execute_activity_method(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ReturnType: ...
 
@@ -1063,6 +1114,7 @@ async def execute_activity_method(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ReturnType: ...
 
@@ -1083,6 +1135,7 @@ async def execute_activity_method(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ReturnType: ...
 
@@ -1102,6 +1155,7 @@ async def execute_activity_method(
     activity_id: str | None = None,
     versioning_intent: VersioningIntent | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> Any:
     """Start an activity from a method and wait for completion.
@@ -1124,6 +1178,7 @@ async def execute_activity_method(
         activity_id=activity_id,
         versioning_intent=versioning_intent,
         summary=summary,
+        event_groups=event_groups,
         priority=priority,
     )
 
@@ -1141,6 +1196,7 @@ class LocalActivityConfig(TypedDict, total=False):
     cancellation_type: ActivityCancellationType
     activity_id: str | None
     summary: str | None
+    event_groups: Sequence[EventGroup] | None
 
 
 # Overload for async no-param activity
@@ -1156,6 +1212,7 @@ def start_local_activity(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ActivityHandle[ReturnType]: ...
 
 
@@ -1172,6 +1229,7 @@ def start_local_activity(
     local_retry_threshold: timedelta | None = None,
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ActivityHandle[ReturnType]: ...
 
 
@@ -1189,6 +1247,7 @@ def start_local_activity(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ActivityHandle[ReturnType]: ...
 
 
@@ -1206,6 +1265,7 @@ def start_local_activity(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ActivityHandle[ReturnType]: ...
 
 
@@ -1223,6 +1283,7 @@ def start_local_activity(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ActivityHandle[ReturnType]: ...
 
 
@@ -1240,6 +1301,7 @@ def start_local_activity(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ActivityHandle[ReturnType]: ...
 
 
@@ -1259,6 +1321,7 @@ def start_local_activity(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ActivityHandle[Any]: ...
 
 
@@ -1276,6 +1339,7 @@ def start_local_activity(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ActivityHandle[Any]:
     """Start a local activity and return its handle.
 
@@ -1304,6 +1368,9 @@ def start_local_activity(
             advanced setting that should not be set unless users are sure they
             need to. Contact Temporal before setting this value.
         summary: Optional summary for the activity.
+        event_groups: Event Groups to associate this command with, in
+            addition to those active in the current scope. See
+            :py:func:`temporalio.workflow.create_event_group`.
 
     Returns:
         An activity handle to the activity which is an async task.
@@ -1320,6 +1387,7 @@ def start_local_activity(
         cancellation_type=cancellation_type,
         activity_id=activity_id,
         summary=summary,
+        event_groups=event_groups,
     )
 
 
@@ -1336,6 +1404,7 @@ async def execute_local_activity(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ReturnType: ...
 
 
@@ -1352,6 +1421,7 @@ async def execute_local_activity(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ReturnType: ...
 
 
@@ -1369,6 +1439,7 @@ async def execute_local_activity(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ReturnType: ...
 
 
@@ -1386,6 +1457,7 @@ async def execute_local_activity(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ReturnType: ...
 
 
@@ -1403,6 +1475,7 @@ async def execute_local_activity(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ReturnType: ...
 
 
@@ -1420,6 +1493,7 @@ async def execute_local_activity(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ReturnType: ...
 
 
@@ -1439,6 +1513,7 @@ async def execute_local_activity(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> Any: ...
 
 
@@ -1456,6 +1531,7 @@ async def execute_local_activity(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> Any:
     """Start a local activity and wait for completion.
 
@@ -1475,6 +1551,7 @@ async def execute_local_activity(
         cancellation_type=cancellation_type,
         activity_id=activity_id,
         summary=summary,
+        event_groups=event_groups,
     )
 
 
@@ -1585,6 +1662,7 @@ def start_local_activity_class(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ActivityHandle[Any]:
     """Start a local activity from a callable class.
 
@@ -1602,6 +1680,7 @@ def start_local_activity_class(
         cancellation_type=cancellation_type,
         activity_id=activity_id,
         summary=summary,
+        event_groups=event_groups,
     )
 
 
@@ -1618,6 +1697,7 @@ async def execute_local_activity_class(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ReturnType: ...
 
 
@@ -1634,6 +1714,7 @@ async def execute_local_activity_class(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ReturnType: ...
 
 
@@ -1651,6 +1732,7 @@ async def execute_local_activity_class(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ReturnType: ...
 
 
@@ -1668,6 +1750,7 @@ async def execute_local_activity_class(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ReturnType: ...
 
 
@@ -1685,6 +1768,7 @@ async def execute_local_activity_class(  # type: ignore[reportOverlappingOverloa
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ReturnType: ...
 
 
@@ -1702,6 +1786,7 @@ async def execute_local_activity_class(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ReturnType: ...
 
 
@@ -1718,6 +1803,7 @@ async def execute_local_activity_class(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> Any:
     """Start a local activity from a callable class and wait for completion.
 
@@ -1737,6 +1823,7 @@ async def execute_local_activity_class(
         cancellation_type=cancellation_type,
         activity_id=activity_id,
         summary=summary,
+        event_groups=event_groups,
     )
 
 
@@ -1753,6 +1840,7 @@ def start_local_activity_method(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ActivityHandle[ReturnType]: ...
 
 
@@ -1769,6 +1857,7 @@ def start_local_activity_method(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ActivityHandle[ReturnType]: ...
 
 
@@ -1786,6 +1875,7 @@ def start_local_activity_method(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ActivityHandle[ReturnType]: ...
 
 
@@ -1803,6 +1893,7 @@ def start_local_activity_method(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ActivityHandle[ReturnType]: ...
 
 
@@ -1820,6 +1911,7 @@ def start_local_activity_method(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ActivityHandle[ReturnType]: ...
 
 
@@ -1837,6 +1929,7 @@ def start_local_activity_method(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ActivityHandle[ReturnType]: ...
 
 
@@ -1853,6 +1946,7 @@ def start_local_activity_method(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ActivityHandle[Any]:
     """Start a local activity from a method.
 
@@ -1870,6 +1964,7 @@ def start_local_activity_method(
         cancellation_type=cancellation_type,
         activity_id=activity_id,
         summary=summary,
+        event_groups=event_groups,
     )
 
 
@@ -1886,6 +1981,7 @@ async def execute_local_activity_method(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ReturnType: ...
 
 
@@ -1902,6 +1998,7 @@ async def execute_local_activity_method(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ReturnType: ...
 
 
@@ -1919,6 +2016,7 @@ async def execute_local_activity_method(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ReturnType: ...
 
 
@@ -1936,6 +2034,7 @@ async def execute_local_activity_method(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ReturnType: ...
 
 
@@ -1953,6 +2052,7 @@ async def execute_local_activity_method(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ReturnType: ...
 
 
@@ -1970,6 +2070,7 @@ async def execute_local_activity_method(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> ReturnType: ...
 
 
@@ -1986,6 +2087,7 @@ async def execute_local_activity_method(
     cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
     activity_id: str | None = None,
     summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> Any:
     """Start a local activity from a method and wait for completion.
 
@@ -2005,4 +2107,5 @@ async def execute_local_activity_method(
         cancellation_type=cancellation_type,
         activity_id=activity_id,
         summary=summary,
+        event_groups=event_groups,
     )
diff --git a/temporalio/workflow/_context.py b/temporalio/workflow/_context.py
index b33f83150..05c17cd30 100644
--- a/temporalio/workflow/_context.py
+++ b/temporalio/workflow/_context.py
@@ -22,6 +22,7 @@
 
 if TYPE_CHECKING:
     from ._activities import ActivityCancellationType, ActivityHandle
+    from ._event_groups import EventGroup
     from ._exceptions import ContinueAsNewVersioningBehavior, VersioningIntent
     from ._nexus import NexusOperationCancellationType, NexusOperationHandle
     from ._workflow_ops import (
@@ -90,6 +91,13 @@ class Info:
     first_execution_run_id: str
     headers: Mapping[str, temporalio.api.common.v1.Payload]
     namespace: str
+
+    original_execution_run_id: str
+    """Run ID recorded on the ``WorkflowExecutionStarted`` event.
+
+    Unlike :py:attr:`run_id`, this value is preserved across workflow resets.
+    """
+
     parent: ParentInfo | None
     root: RootInfo | None
     priority: temporalio.common.Priority
@@ -296,6 +304,7 @@ def workflow_continue_as_new(
         ),
         versioning_intent: VersioningIntent | None,
         initial_versioning_behavior: ContinueAsNewVersioningBehavior | None,
+        event_groups: Sequence[EventGroup] | None = None,
     ) -> NoReturn: ...
 
     @abstractmethod
@@ -413,6 +422,7 @@ def workflow_start_activity(
         activity_id: str | None,
         versioning_intent: VersioningIntent | None,
         summary: str | None = None,
+        event_groups: Sequence[EventGroup] | None = None,
         priority: temporalio.common.Priority = temporalio.common.Priority.default,
     ) -> ActivityHandle[Any]: ...
 
@@ -440,6 +450,7 @@ async def workflow_start_child_workflow(
         versioning_intent: VersioningIntent | None,
         static_summary: str | None = None,
         static_details: str | None = None,
+        event_groups: Sequence[EventGroup] | None = None,
         priority: temporalio.common.Priority = temporalio.common.Priority.default,
     ) -> ChildWorkflowHandle[Any, Any]: ...
 
@@ -457,6 +468,7 @@ def workflow_start_local_activity(
         cancellation_type: ActivityCancellationType,
         activity_id: str | None,
         summary: str | None,
+        event_groups: Sequence[EventGroup] | None = None,
     ) -> ActivityHandle[Any]: ...
 
     @abstractmethod
@@ -473,6 +485,7 @@ async def workflow_start_nexus_operation(
         cancellation_type: NexusOperationCancellationType,
         headers: Mapping[str, str] | None,
         summary: str | None,
+        event_groups: Sequence[EventGroup] | None = None,
     ) -> NexusOperationHandle[OutputT]: ...
 
     @abstractmethod
@@ -489,7 +502,11 @@ def workflow_upsert_search_attributes(
 
     @abstractmethod
     async def workflow_sleep(
-        self, duration: float, *, summary: str | None = None
+        self,
+        duration: float,
+        *,
+        summary: str | None = None,
+        event_groups: Sequence[EventGroup] | None = None,
     ) -> None: ...
 
     @abstractmethod
@@ -499,6 +516,7 @@ async def workflow_wait_condition(
         *,
         timeout: float | None = None,
         timeout_summary: str | None = None,
+        event_groups: Sequence[EventGroup] | None = None,
     ) -> None: ...
 
     @abstractmethod
@@ -929,19 +947,28 @@ def uuid7() -> uuid.UUID:
     )
 
 
-async def sleep(duration: float | timedelta, *, summary: str | None = None) -> None:
+async def sleep(
+    duration: float | timedelta,
+    *,
+    summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
+) -> None:
     """Sleep for the given duration.
 
     Args:
         duration: Duration to sleep in seconds or as a timedelta.
         summary: A single-line fixed summary for this timer that may appear in UI/CLI.
             This can be in single-line Temporal markdown format.
+        event_groups: Event Groups to associate this command with, in addition
+            to those active in the current scope. See
+            :py:func:`temporalio.workflow.create_event_group`.
     """
     await _Runtime.current().workflow_sleep(
         duration=(
             duration.total_seconds() if isinstance(duration, timedelta) else duration
         ),
         summary=summary,
+        event_groups=event_groups,
     )
 
 
@@ -950,6 +977,7 @@ async def wait_condition(
     *,
     timeout: timedelta | float | None = None,
     timeout_summary: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> None:
     """Wait on a callback to become true.
 
@@ -968,9 +996,14 @@ async def wait_condition(
         timeout_summary: Optional simple string identifying the timer (created if ``timeout`` is
             present) that may be visible in UI/CLI. While it can be normal text, it is best to treat
             as a timer ID.
+        event_groups: Event Groups to associate the timer command (created if
+            ``timeout`` is present) with, in addition to those active in the
+            current scope. See
+            :py:func:`temporalio.workflow.create_event_group`.
     """
     await _Runtime.current().workflow_wait_condition(
         fn,
         timeout=timeout.total_seconds() if isinstance(timeout, timedelta) else timeout,
         timeout_summary=timeout_summary,
+        event_groups=event_groups,
     )
diff --git a/temporalio/workflow/_event_groups.py b/temporalio/workflow/_event_groups.py
new file mode 100644
index 000000000..cd24f91c4
--- /dev/null
+++ b/temporalio/workflow/_event_groups.py
@@ -0,0 +1,242 @@
+"""Event Groups, a way to regroup logically related workflow events.
+
+.. warning::
+    Event Groups is an experimental API and may change without notice.
+"""
+
+from __future__ import annotations
+
+import contextvars
+import hashlib
+from abc import ABC, abstractmethod
+from collections.abc import Iterator, Sequence
+from contextlib import contextmanager
+from dataclasses import dataclass
+
+import temporalio.api.sdk.v1
+import temporalio.converter
+
+from ._context import _Runtime
+
+__all__ = [
+    "EventGroup",
+    "create_event_group",
+]
+
+
+class EventGroup(ABC):
+    """A discrete token associating workflow commands, and the history events
+    they produce, with a logical group for UI and observability purposes.
+
+    Multiple Event Groups may be attached to a single command, and a single
+    Event Group may be attached to multiple commands.
+
+    Instances are created with :py:func:`create_event_group`. They may be
+    attached to specific commands via the ``event_groups`` option of the API
+    producing the command, or to every command produced within a block of
+    workflow code via :py:meth:`scope`.
+
+    .. warning::
+        Event Groups is an experimental API and may change without notice.
+    """
+
+    @contextmanager
+    def scope(self) -> Iterator[None]:
+        """Context manager attaching this Event Group to every command produced
+        within it.
+
+        Scopes nest and compose: a command produced inside an inner scope
+        carries the Event Groups of all enclosing scopes. Coroutines started
+        within a scope inherit it, since they capture the context active at
+        their creation.
+
+        Only usable from within a workflow.
+
+        .. warning::
+            Event Groups is an experimental API and may change without notice.
+        """
+        _Runtime.current()
+        token = _active_event_groups.set(self._applied_over(_active_event_groups.get()))
+        try:
+            yield
+        finally:
+            try:
+                _active_event_groups.reset(token)
+            except ValueError:
+                # Unwinding from a context other than the one the scope was
+                # entered in, which happens when a coroutine suspended inside
+                # the scope is closed rather than resumed. The context the
+                # value was set in is being discarded anyway.
+                pass
+
+    @abstractmethod
+    def _applied_over(self, active: _ActiveEventGroups) -> _ActiveEventGroups:
+        """Return the active set resulting from entering this group's scope."""
+        ...
+
+    @abstractmethod
+    def _to_proto(self) -> temporalio.api.sdk.v1.EventGroupMarker:
+        """Serialize as the marker attached to a workflow command."""
+        ...
+
+
+class _LabelEventGroup(EventGroup):
+    """An Event Group explicitly created by workflow code."""
+
+    def __init__(self, id: str, label: str) -> None:
+        self._id = id
+        self._label = label
+
+    def _applied_over(self, active: _ActiveEventGroups) -> _ActiveEventGroups:
+        return _ActiveEventGroups(
+            implicit=active.implicit,
+            explicit=_with_group(active.explicit, self),
+        )
+
+    def _to_proto(self) -> temporalio.api.sdk.v1.EventGroupMarker:
+        # Deliberately the SDK's default converter rather than the worker's own: the UI and CLI
+        # rely on the label being a json/plain string, which a user-provided converter could
+        # break.
+        return temporalio.api.sdk.v1.EventGroupMarker(
+            label=temporalio.api.sdk.v1.EventGroupMarker.Label(
+                id=self._id,
+                label=temporalio.converter.PayloadConverter.default.to_payload(
+                    self._label
+                ),
+            )
+        )
+
+
+class _ImplicitEventGroup(EventGroup):
+    """An Event Group created by the SDK around an inbound signal or update.
+
+    The workflow's main function deliberately gets no such group, so commands it
+    produces outside any explicit scope carry no markers at all.
+    """
+
+    def __init__(self, marker: temporalio.api.sdk.v1.EventGroupMarker) -> None:
+        self._marker = marker
+
+    def _applied_over(self, active: _ActiveEventGroups) -> _ActiveEventGroups:
+        # Implicit groups intentionally do not inherit the enclosing scope: a
+        # handler registered inside an explicit scope must not attribute its
+        # commands to that scope.
+        return _ActiveEventGroups(implicit=self)
+
+    def _to_proto(self) -> temporalio.api.sdk.v1.EventGroupMarker:
+        return self._marker
+
+
+@dataclass(frozen=True)
+class _ActiveEventGroups:
+    implicit: EventGroup | None = None
+    explicit: tuple[_LabelEventGroup, ...] = ()
+
+
+_active_event_groups: contextvars.ContextVar[_ActiveEventGroups] = (
+    contextvars.ContextVar(
+        "__temporal_active_event_groups", default=_ActiveEventGroups()
+    )
+)
+
+
+def create_event_group(label: str, *, id: str | None = None) -> EventGroup:
+    """Create an Event Group that can be attached to commands produced by this
+    workflow.
+
+    Args:
+        label: User-visible label for the group, surfaced in the UI and CLI.
+            The label is converted to a payload using the SDK's default payload
+            converter, not the one configured on the worker, then encoded using
+            the worker's configured payload codecs.
+
+            Note that when no ``id`` is given, the id is derived from the label
+            using a hash function. Given short and predictable labels,
+            brute-forcing the hashed value may be computationally feasible,
+            thereby recovering the label. Avoid putting sensitive information
+            in labels, or provide an explicit ``id``.
+        id: Opaque identifier determining whether two Event Groups are the
+            same. Events are grouped together if and only if their groups have
+            the same id, without regard to their labels; only the first label
+            seen for a given id is used. Defaults to a deterministic,
+            replay-stable value derived from the label. The id is not encoded
+            using payload codecs.
+
+    Returns:
+        The new Event Group.
+
+    .. warning::
+        Event Groups is an experimental API and may change without notice.
+    """
+    info = _Runtime.current().workflow_info()
+    if not label:
+        raise ValueError("Event group label cannot be empty")
+    if id is None:
+        # Salted with the run id so that the label cannot be recovered from the
+        # id using precomputed hashes. This is the run id of the
+        # WorkflowExecutionStarted event, which is preserved across resets, so
+        # ids remain stable on replay and after a reset.
+        id = hashlib.sha1(
+            f"{info.original_execution_run_id}{label}".encode()
+        ).hexdigest()
+    elif not id:
+        raise ValueError("Event group id cannot be empty")
+    return _LabelEventGroup(id, label)
+
+
+def _inbound_event_group(event_id: int) -> EventGroup:
+    """Create the implicit Event Group for an inbound signal's history event."""
+    if event_id <= 0:
+        raise ValueError(f"Invalid inbound event id: {event_id}")
+    return _ImplicitEventGroup(
+        temporalio.api.sdk.v1.EventGroupMarker(
+            inbound_event=temporalio.api.sdk.v1.EventGroupMarker.InboundEvent(
+                inbound_event_id=event_id
+            )
+        )
+    )
+
+
+def _inbound_update_event_group(update_id: str) -> EventGroup:
+    """Create the implicit Event Group for an inbound update."""
+    return _ImplicitEventGroup(
+        temporalio.api.sdk.v1.EventGroupMarker(
+            inbound_update=temporalio.api.sdk.v1.EventGroupMarker.InboundUpdate(
+                inbound_update_id=update_id
+            )
+        )
+    )
+
+
+def _event_group_markers_to_proto(
+    event_groups: Sequence[EventGroup] | None,
+) -> list[temporalio.api.sdk.v1.EventGroupMarker]:
+    """Merge the given Event Groups with those active in the current scope and
+    serialize them as the markers attached to a workflow command.
+
+    Must be called from the context the command was requested in, which is not
+    necessarily the one it is ultimately built in.
+    """
+    active = _active_event_groups.get()
+    explicit = active.explicit
+    for group in event_groups or ():
+        if not isinstance(group, _LabelEventGroup):
+            raise TypeError(
+                "Event groups must be created with workflow.create_event_group()"
+            )
+        explicit = _with_group(explicit, group)
+    groups: list[EventGroup] = list(explicit)
+    if active.implicit:
+        groups.insert(0, active.implicit)
+    return [group._to_proto() for group in groups]
+
+
+def _with_group(
+    groups: tuple[_LabelEventGroup, ...], group: _LabelEventGroup
+) -> tuple[_LabelEventGroup, ...]:
+    """Add a group to a set of groups, deduplicating by id."""
+    if any(existing._id == group._id for existing in groups):
+        return tuple(
+            group if existing._id == group._id else existing for existing in groups
+        )
+    return (*groups, group)
diff --git a/temporalio/workflow/_nexus.py b/temporalio/workflow/_nexus.py
index 29bd10715..d8ad124cc 100644
--- a/temporalio/workflow/_nexus.py
+++ b/temporalio/workflow/_nexus.py
@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from collections.abc import Awaitable, Callable, Generator, Mapping
+from collections.abc import Awaitable, Callable, Generator, Mapping, Sequence
 from datetime import timedelta
 from enum import IntEnum
 from typing import Any, Generic, overload
@@ -15,6 +15,7 @@
 from temporalio.types import NexusServiceType
 
 from ._context import _Runtime
+from ._event_groups import EventGroup
 
 __all__ = [
     "NexusClient",
@@ -112,6 +113,7 @@ async def start_operation(
         cancellation_type: NexusOperationCancellationType = NexusOperationCancellationType.WAIT_COMPLETED,
         headers: Mapping[str, str] | None = None,
         summary: str | None = None,
+        event_groups: Sequence[EventGroup] | None = None,
     ) -> NexusOperationHandle[OutputT]: ...
 
     # Overload for string operation name
@@ -129,6 +131,7 @@ async def start_operation(
         cancellation_type: NexusOperationCancellationType = NexusOperationCancellationType.WAIT_COMPLETED,
         headers: Mapping[str, str] | None = None,
         summary: str | None = None,
+        event_groups: Sequence[EventGroup] | None = None,
     ) -> NexusOperationHandle[OutputT]: ...
 
     # Overload for workflow_run_operation methods
@@ -149,6 +152,7 @@ async def start_operation(
         cancellation_type: NexusOperationCancellationType = NexusOperationCancellationType.WAIT_COMPLETED,
         headers: Mapping[str, str] | None = None,
         summary: str | None = None,
+        event_groups: Sequence[EventGroup] | None = None,
     ) -> NexusOperationHandle[OutputT]: ...
 
     # Overload for sync_operation methods (async def)
@@ -169,6 +173,7 @@ async def start_operation(
         cancellation_type: NexusOperationCancellationType = NexusOperationCancellationType.WAIT_COMPLETED,
         headers: Mapping[str, str] | None = None,
         summary: str | None = None,
+        event_groups: Sequence[EventGroup] | None = None,
     ) -> NexusOperationHandle[OutputT]: ...
 
     # Overload for sync_operation methods (def)
@@ -189,6 +194,7 @@ async def start_operation(
         cancellation_type: NexusOperationCancellationType = NexusOperationCancellationType.WAIT_COMPLETED,
         headers: Mapping[str, str] | None = None,
         summary: str | None = None,
+        event_groups: Sequence[EventGroup] | None = None,
     ) -> NexusOperationHandle[OutputT]: ...
 
     # Overload for operation_handler
@@ -208,6 +214,7 @@ async def start_operation(
         cancellation_type: NexusOperationCancellationType = NexusOperationCancellationType.WAIT_COMPLETED,
         headers: Mapping[str, str] | None = None,
         summary: str | None = None,
+        event_groups: Sequence[EventGroup] | None = None,
     ) -> NexusOperationHandle[OutputT]: ...
 
     # Overload for temporal_operation methods
@@ -233,6 +240,7 @@ async def start_operation(
         cancellation_type: NexusOperationCancellationType = NexusOperationCancellationType.WAIT_COMPLETED,
         headers: Mapping[str, str] | None = None,
         summary: str | None = None,
+        event_groups: Sequence[EventGroup] | None = None,
     ) -> NexusOperationHandle[OutputT]: ...
 
     @abstractmethod
@@ -248,6 +256,7 @@ async def start_operation(
         cancellation_type: NexusOperationCancellationType = NexusOperationCancellationType.WAIT_COMPLETED,
         headers: Mapping[str, str] | None = None,
         summary: str | None = None,
+        event_groups: Sequence[EventGroup] | None = None,
     ) -> Any:
         """Start a Nexus operation and return its handle.
 
@@ -283,6 +292,7 @@ async def execute_operation(
         cancellation_type: NexusOperationCancellationType = NexusOperationCancellationType.WAIT_COMPLETED,
         headers: Mapping[str, str] | None = None,
         summary: str | None = None,
+        event_groups: Sequence[EventGroup] | None = None,
     ) -> OutputT: ...
 
     # Overload for string operation name
@@ -300,6 +310,7 @@ async def execute_operation(
         cancellation_type: NexusOperationCancellationType = NexusOperationCancellationType.WAIT_COMPLETED,
         headers: Mapping[str, str] | None = None,
         summary: str | None = None,
+        event_groups: Sequence[EventGroup] | None = None,
     ) -> OutputT: ...
 
     # Overload for workflow_run_operation methods
@@ -320,6 +331,7 @@ async def execute_operation(
         cancellation_type: NexusOperationCancellationType = NexusOperationCancellationType.WAIT_COMPLETED,
         headers: Mapping[str, str] | None = None,
         summary: str | None = None,
+        event_groups: Sequence[EventGroup] | None = None,
     ) -> OutputT: ...
 
     # Overload for sync_operation methods (async def)
@@ -340,6 +352,7 @@ async def execute_operation(
         cancellation_type: NexusOperationCancellationType = NexusOperationCancellationType.WAIT_COMPLETED,
         headers: Mapping[str, str] | None = None,
         summary: str | None = None,
+        event_groups: Sequence[EventGroup] | None = None,
     ) -> OutputT: ...
 
     # Overload for sync_operation methods (def)
@@ -360,6 +373,7 @@ async def execute_operation(
         cancellation_type: NexusOperationCancellationType = NexusOperationCancellationType.WAIT_COMPLETED,
         headers: Mapping[str, str] | None = None,
         summary: str | None = None,
+        event_groups: Sequence[EventGroup] | None = None,
     ) -> OutputT: ...
 
     # Overload for operation_handler
@@ -380,6 +394,7 @@ async def execute_operation(
         cancellation_type: NexusOperationCancellationType = NexusOperationCancellationType.WAIT_COMPLETED,
         headers: Mapping[str, str] | None = None,
         summary: str | None = None,
+        event_groups: Sequence[EventGroup] | None = None,
     ) -> OutputT: ...
 
     # Overload for temporal_operation methods
@@ -405,6 +420,7 @@ async def execute_operation(
         cancellation_type: NexusOperationCancellationType = NexusOperationCancellationType.WAIT_COMPLETED,
         headers: Mapping[str, str] | None = None,
         summary: str | None = None,
+        event_groups: Sequence[EventGroup] | None = None,
     ) -> OutputT: ...
 
     @abstractmethod
@@ -420,6 +436,7 @@ async def execute_operation(
         cancellation_type: NexusOperationCancellationType = NexusOperationCancellationType.WAIT_COMPLETED,
         headers: Mapping[str, str] | None = None,
         summary: str | None = None,
+        event_groups: Sequence[EventGroup] | None = None,
     ) -> Any:
         """Execute a Nexus operation and return its result.
 
@@ -477,6 +494,7 @@ async def start_operation(
         cancellation_type: NexusOperationCancellationType = NexusOperationCancellationType.WAIT_COMPLETED,
         headers: Mapping[str, str] | None = None,
         summary: str | None = None,
+        event_groups: Sequence[EventGroup] | None = None,
     ) -> Any:
         return await _Runtime.current().workflow_start_nexus_operation(
             endpoint=self.endpoint,
@@ -490,6 +508,7 @@ async def start_operation(
             cancellation_type=cancellation_type,
             headers=headers,
             summary=summary,
+            event_groups=event_groups,
         )
 
     async def execute_operation(
@@ -504,6 +523,7 @@ async def execute_operation(
         cancellation_type: NexusOperationCancellationType = NexusOperationCancellationType.WAIT_COMPLETED,
         headers: Mapping[str, str] | None = None,
         summary: str | None = None,
+        event_groups: Sequence[EventGroup] | None = None,
     ) -> Any:
         handle = await self.start_operation(
             operation,
@@ -515,6 +535,7 @@ async def execute_operation(
             cancellation_type=cancellation_type,
             headers=headers,
             summary=summary,
+            event_groups=event_groups,
         )
         return await handle
 
diff --git a/temporalio/workflow/_workflow_ops.py b/temporalio/workflow/_workflow_ops.py
index f80ca1bdb..5e0dfc14b 100644
--- a/temporalio/workflow/_workflow_ops.py
+++ b/temporalio/workflow/_workflow_ops.py
@@ -20,6 +20,7 @@
 )
 from ._activities import _AsyncioTask
 from ._context import _Runtime, uuid4
+from ._event_groups import EventGroup
 from ._exceptions import ContinueAsNewVersioningBehavior, VersioningIntent
 
 __all__ = [
@@ -172,6 +173,7 @@ class ChildWorkflowConfig(TypedDict, total=False):
     versioning_intent: VersioningIntent | None
     static_summary: str | None
     static_details: str | None
+    event_groups: Sequence[EventGroup] | None
     priority: temporalio.common.Priority
 
 
@@ -198,6 +200,7 @@ async def start_child_workflow(
     versioning_intent: VersioningIntent | None = None,
     static_summary: str | None = None,
     static_details: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ChildWorkflowHandle[SelfType, ReturnType]: ...
 
@@ -226,6 +229,7 @@ async def start_child_workflow(
     versioning_intent: VersioningIntent | None = None,
     static_summary: str | None = None,
     static_details: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ChildWorkflowHandle[SelfType, ReturnType]: ...
 
@@ -254,6 +258,7 @@ async def start_child_workflow(
     versioning_intent: VersioningIntent | None = None,
     static_summary: str | None = None,
     static_details: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ChildWorkflowHandle[SelfType, ReturnType]: ...
 
@@ -284,6 +289,7 @@ async def start_child_workflow(
     versioning_intent: VersioningIntent | None = None,
     static_summary: str | None = None,
     static_details: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ChildWorkflowHandle[Any, Any]: ...
 
@@ -312,6 +318,7 @@ async def start_child_workflow(
     versioning_intent: VersioningIntent | None = None,
     static_summary: str | None = None,
     static_details: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ChildWorkflowHandle[Any, Any]:
     """Start a child workflow and return its handle.
@@ -374,6 +381,7 @@ async def start_child_workflow(
         versioning_intent=versioning_intent,
         static_summary=static_summary,
         static_details=static_details,
+        event_groups=event_groups,
         priority=priority,
     )
 
@@ -401,6 +409,7 @@ async def execute_child_workflow(
     versioning_intent: VersioningIntent | None = None,
     static_summary: str | None = None,
     static_details: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ReturnType: ...
 
@@ -429,6 +438,7 @@ async def execute_child_workflow(
     versioning_intent: VersioningIntent | None = None,
     static_summary: str | None = None,
     static_details: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ReturnType: ...
 
@@ -457,6 +467,7 @@ async def execute_child_workflow(
     versioning_intent: VersioningIntent | None = None,
     static_summary: str | None = None,
     static_details: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> ReturnType: ...
 
@@ -487,6 +498,7 @@ async def execute_child_workflow(
     versioning_intent: VersioningIntent | None = None,
     static_summary: str | None = None,
     static_details: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> Any: ...
 
@@ -515,6 +527,7 @@ async def execute_child_workflow(
     versioning_intent: VersioningIntent | None = None,
     static_summary: str | None = None,
     static_details: str | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
     priority: temporalio.common.Priority = temporalio.common.Priority.default,
 ) -> Any:
     """Start a child workflow and wait for completion.
@@ -543,6 +556,7 @@ async def execute_child_workflow(
         versioning_intent=versioning_intent,
         static_summary=static_summary,
         static_details=static_details,
+        event_groups=event_groups,
         priority=priority,
     )
     return await handle
@@ -692,6 +706,7 @@ def continue_as_new(
     ) = None,
     versioning_intent: VersioningIntent | None = None,
     initial_versioning_behavior: ContinueAsNewVersioningBehavior | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> NoReturn: ...
 
 
@@ -712,6 +727,7 @@ def continue_as_new(
     ) = None,
     versioning_intent: VersioningIntent | None = None,
     initial_versioning_behavior: ContinueAsNewVersioningBehavior | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> NoReturn: ...
 
 
@@ -733,6 +749,7 @@ def continue_as_new(
     ) = None,
     versioning_intent: VersioningIntent | None = None,
     initial_versioning_behavior: ContinueAsNewVersioningBehavior | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> NoReturn: ...
 
 
@@ -754,6 +771,7 @@ def continue_as_new(
     ) = None,
     versioning_intent: VersioningIntent | None = None,
     initial_versioning_behavior: ContinueAsNewVersioningBehavior | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> NoReturn: ...
 
 
@@ -775,6 +793,7 @@ def continue_as_new(
     ) = None,
     versioning_intent: VersioningIntent | None = None,
     initial_versioning_behavior: ContinueAsNewVersioningBehavior | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> NoReturn: ...
 
 
@@ -795,6 +814,7 @@ def continue_as_new(
     ) = None,
     versioning_intent: VersioningIntent | None = None,
     initial_versioning_behavior: ContinueAsNewVersioningBehavior | None = None,
+    event_groups: Sequence[EventGroup] | None = None,
 ) -> NoReturn:
     """Stop the workflow immediately and continue as new.
 
@@ -840,6 +860,7 @@ def continue_as_new(
         search_attributes=search_attributes,
         versioning_intent=versioning_intent,
         initial_versioning_behavior=initial_versioning_behavior,
+        event_groups=event_groups,
     )
 
 
diff --git a/tests/test_workflow_exports.py b/tests/test_workflow_exports.py
index 8788addc5..400ebdc27 100644
--- a/tests/test_workflow_exports.py
+++ b/tests/test_workflow_exports.py
@@ -21,6 +21,7 @@
     "ContinueAsNewError",
     "ContinueAsNewVersioningBehavior",
     "DynamicWorkflowConfig",
+    "EventGroup",
     "ExternalWorkflowHandle",
     "HandlerUnfinishedPolicy",
     "Info",
@@ -64,7 +65,10 @@
     "_bind_method",
     "_build_log_context",
     "_current_update_info",
+    "_event_group_markers_to_proto",
     "_imports_passed_through",
+    "_inbound_event_group",
+    "_inbound_update_event_group",
     "_in_sandbox",
     "_is_unbound_method_on_cls",
     "_parameters_identical_up_to_naming",
@@ -78,6 +82,7 @@
     "annotations",
     "as_completed",
     "continue_as_new",
+    "create_event_group",
     "create_nexus_client",
     "current_update_info",
     "defn",
diff --git a/tests/worker/test_event_groups.py b/tests/worker/test_event_groups.py
new file mode 100644
index 000000000..181a71359
--- /dev/null
+++ b/tests/worker/test_event_groups.py
@@ -0,0 +1,1932 @@
+"""Event Groups tests, following Workspace/test-plan.md.
+
+The existing ``test_event_groups.py`` is the pre-plan smoke suite and is left
+in place until this file replaces it. Case IDs in comments are the plan's;
+TypeScript's suite is a style reference only and may still change.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import uuid
+from collections.abc import Sequence
+from datetime import timedelta
+
+import nexusrpc
+import pytest
+
+from temporalio import activity, workflow
+from temporalio.api.common.v1 import Payload, WorkflowExecution
+from temporalio.api.enums.v1 import EventType
+from temporalio.api.history.v1 import HistoryEvent
+from temporalio.api.sdk.v1 import EventGroupMarker
+from temporalio.api.workflowservice.v1 import ResetWorkflowExecutionRequest
+from temporalio.client import Client, WorkflowHandle
+from temporalio.common import RawValue, RetryPolicy, SearchAttributeKey
+from temporalio.converter import (
+    CompositePayloadConverter,
+    DataConverter,
+    DefaultPayloadConverter,
+    EncodingPayloadConverter,
+    PayloadCodec,
+    PayloadConverter,
+)
+from temporalio.exceptions import (
+    ActivityError,
+    ApplicationError,
+    ChildWorkflowError,
+    NexusOperationError,
+    TemporalError,
+)
+from temporalio.testing import WorkflowEnvironment
+from temporalio.worker import Worker
+from tests.helpers import assert_eventually, ensure_search_attributes_present, new_worker
+from tests.helpers.nexus import make_nexus_endpoint_name
+
+# These tests need a server that transcribes Event Group markers onto history.
+# Time-skipping (the Java test server) does not.
+
+_ACT_TIMEOUT = timedelta(seconds=10)
+
+
+def _require_event_groups_server(env: WorkflowEnvironment) -> None:
+    if env.supports_time_skipping:
+        pytest.skip("Event Groups require a server that transcribes markers")
+
+
+####################################################################################################
+# 1. Explicit Event Groups Marker Label IDs (`EG-LABEL-ID`)
+####################################################################################################
+
+
+@workflow.defn
+class DerivedIdsWorkflow:
+    @workflow.run
+    async def run(self) -> None:
+        a = workflow.create_event_group("aaa")
+        b1 = workflow.create_event_group("bbb")
+        b2 = workflow.create_event_group("bbb")
+        await _activity("activity-a", [a])
+        await _activity("activity-b1", [b1])
+        await _activity("activity-b2", [b2])
+
+
+async def test_derived_label_ids(client: Client, env: WorkflowEnvironment):
+    _require_event_groups_server(env)
+
+    async with new_worker(
+        client, DerivedIdsWorkflow, activities=[noop_activity]
+    ) as worker:
+        handle1 = await client.start_workflow(
+            DerivedIdsWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        handle2 = await client.start_workflow(
+            DerivedIdsWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle1.result()
+        await handle2.result()
+        events1 = await _fetch_events(handle1)
+        events2 = await _fetch_events(handle2)
+        run_id1 = _run_id(handle1)
+
+        assert (
+            len(_events_of_type(events1, EventType.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED))
+            == 3
+        )
+        assert (
+            len(_events_of_type(events2, EventType.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED))
+            == 3
+        )
+
+        # EG-LABEL-ID-00: Derived IDs match SHA1(original_execution_run_id + label)
+        _assert_marker_ids(
+            _activity_event(events1, "activity-a"),
+            _label_marker_id(_default_marker_id(run_id1, "aaa")),
+        )
+
+        # EG-LABEL-ID-01: same label + no user-provided ID => same group
+        assert _marker_ids(_activity_event(events1, "activity-b1")) == _marker_ids(
+            _activity_event(events1, "activity-b2")
+        )
+
+        # EG-LABEL-ID-02: different labels + no user-provided ID => distinct groups
+        assert _marker_ids(_activity_event(events1, "activity-a")) != _marker_ids(
+            _activity_event(events1, "activity-b1")
+        )
+
+        # EG-LABEL-ID-03: same labels + different workflow execs => distinct groups
+        assert _marker_ids(_activity_event(events1, "activity-a")) != _marker_ids(
+            _activity_event(events2, "activity-a")
+        )
+
+
+async def test_derived_label_ids_stable_across_reset(
+    client: Client, env: WorkflowEnvironment
+):
+    _require_event_groups_server(env)
+
+    async with new_worker(
+        client, DerivedIdsWorkflow, activities=[noop_activity]
+    ) as worker:
+        handle = await client.start_workflow(
+            DerivedIdsWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle.result()
+        events = await _fetch_events(handle)
+        original_run_id = _run_id(handle)
+
+        first_wft_started = next(
+            e.event_id
+            for e in events
+            if e.event_type == EventType.EVENT_TYPE_WORKFLOW_TASK_STARTED
+        )
+        reset = await client.workflow_service.reset_workflow_execution(
+            ResetWorkflowExecutionRequest(
+                namespace=client.namespace,
+                workflow_execution=WorkflowExecution(
+                    workflow_id=handle.id, run_id=original_run_id
+                ),
+                reason="test event group id stability across reset",
+                request_id=str(uuid.uuid4()),
+                workflow_task_finish_event_id=first_wft_started,
+            )
+        )
+        assert reset.run_id != original_run_id
+        reset_handle = client.get_workflow_handle(handle.id, run_id=reset.run_id)
+        await reset_handle.result()
+        reset_events = await _fetch_events(reset_handle)
+
+        assert (
+            len(_events_of_type(events, EventType.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED))
+            == 3
+        )
+        assert (
+            len(
+                _events_of_type(
+                    reset_events, EventType.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED
+                )
+            )
+            == 3
+        )
+
+        # Control: reset re-executed the first workflow task
+        assert (
+            _events_of_type(events, EventType.EVENT_TYPE_WORKFLOW_TASK_COMPLETED)[
+                0
+            ].event_time
+            != _events_of_type(
+                reset_events, EventType.EVENT_TYPE_WORKFLOW_TASK_COMPLETED
+            )[0].event_time
+        )
+
+        # EG-LABEL-ID-04: derived IDs are based on the original execution run id
+        _assert_marker_ids(
+            _activity_event(reset_events, "activity-a"),
+            _label_marker_id(_default_marker_id(original_run_id, "aaa")),
+        )
+        assert _marker_ids(_activity_event(events, "activity-b1")) == _marker_ids(
+            _activity_event(reset_events, "activity-b1")
+        )
+
+
+@workflow.defn
+class UserProvidedIdsWorkflow:
+    @workflow.run
+    async def run(self) -> None:
+        c = workflow.create_event_group("ccc", id="c-id")
+        d1 = workflow.create_event_group("ddd1", id="d-id")
+        d2 = workflow.create_event_group("ddd2", id="d-id")
+        await _activity("activity-c", [c])
+        await _activity("activity-d1", [d1])
+        await _activity("activity-d2", [d2])
+
+
+async def test_user_provided_label_ids(client: Client, env: WorkflowEnvironment):
+    _require_event_groups_server(env)
+
+    async with new_worker(
+        client, UserProvidedIdsWorkflow, activities=[noop_activity]
+    ) as worker:
+        handle = await client.start_workflow(
+            UserProvidedIdsWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle.result()
+        events = await _fetch_events(handle)
+        assert (
+            len(_events_of_type(events, EventType.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED))
+            == 3
+        )
+
+        # EG-LABEL-ID-20: user-provided IDs are used verbatim
+        _assert_marker_ids(
+            _activity_event(events, "activity-c"), _label_marker_id("c-id")
+        )
+
+        # EG-LABEL-ID-21: different labels + same user-provided ID => same group
+        assert _marker_ids(_activity_event(events, "activity-d1")) == _marker_ids(
+            _activity_event(events, "activity-d2")
+        )
+
+
+####################################################################################################
+# 2. Explicit Event Groups Marker Label Payload (`EG-LABEL-PAYLOAD`)
+####################################################################################################
+
+
+@workflow.defn
+class LabelPayloadWorkflow:
+    @workflow.run
+    async def run(self) -> None:
+        a = workflow.create_event_group("aaa")
+        b = workflow.create_event_group("bbb", id="b-id")
+        # Control: activity arguments go through the worker's payload converter, so this is how the
+        # custom-converter test proves that converter is actually installed.
+        await workflow.execute_activity(
+            control_activity,
+            "control",
+            start_to_close_timeout=_ACT_TIMEOUT,
+            activity_id="control",
+        )
+        await _activity("activity-a", [a])
+        await _activity("activity-b", [b])
+
+
+class _CustomStringConverter(EncodingPayloadConverter):
+    @property
+    def encoding(self) -> str:
+        return "custom"
+
+    def to_payload(self, value: object) -> Payload | None:
+        if isinstance(value, str):
+            return Payload(
+                metadata={"encoding": b"custom"},
+                data=f"custom-converter-{value}".encode(),
+            )
+        return None
+
+    def from_payload(self, payload: Payload, type_hint: type | None = None) -> str:
+        text = payload.data.decode()
+        prefix = "custom-converter-"
+        return text[len(prefix) :] if text.startswith(prefix) else text
+
+
+class _CustomPayloadConverter(CompositePayloadConverter):
+    def __init__(self) -> None:
+        super().__init__(
+            _CustomStringConverter(),
+            *DefaultPayloadConverter.default_encoding_payload_converters,
+        )
+
+
+class _WrappingPayloadCodec(PayloadCodec):
+    async def encode(self, payloads: Sequence[Payload]) -> list[Payload]:
+        return [
+            Payload(
+                metadata={"encoding": b"binary/wrapped"}, data=p.SerializeToString()
+            )
+            for p in payloads
+        ]
+
+    async def decode(self, payloads: Sequence[Payload]) -> list[Payload]:
+        decoded: list[Payload] = []
+        for payload in payloads:
+            inner = Payload()
+            inner.ParseFromString(payload.data)
+            decoded.append(inner)
+        return decoded
+
+
+async def test_label_payload_is_json_plain(client: Client, env: WorkflowEnvironment):
+    _require_event_groups_server(env)
+
+    async with new_worker(
+        client,
+        LabelPayloadWorkflow,
+        activities=[noop_activity, control_activity],
+    ) as worker:
+        handle = await client.start_workflow(
+            LabelPayloadWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle.result()
+        events = await _fetch_events(handle)
+        run_id = _run_id(handle)
+        a_id = _default_marker_id(run_id, "aaa")
+
+        activity_a = _activity_event(events, "activity-a")
+        activity_b = _activity_event(events, "activity-b")
+        _assert_markers(activity_a, _label_marker(a_id, "aaa"))
+        _assert_markers(activity_b, _label_marker("b-id", "bbb"))
+
+        # EG-LABEL-PAYLOAD-00: label payload is a json/plain JSON string
+        assert _label_payload_of(activity_a, a_id) == ("json/plain", '"aaa"')
+        assert _label_payload_of(activity_b, "b-id") == ("json/plain", '"bbb"')
+
+
+async def test_label_payload_uses_default_converter_not_worker_converter(
+    env: WorkflowEnvironment,
+):
+    _require_event_groups_server(env)
+
+    custom_client = await env.connect_client(
+        data_converter=DataConverter(payload_converter_class=_CustomPayloadConverter)
+    )
+    async with new_worker(
+        custom_client,
+        LabelPayloadWorkflow,
+        activities=[noop_activity, control_activity],
+    ) as worker:
+        handle = await custom_client.start_workflow(
+            LabelPayloadWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle.result()
+        events = await _fetch_events(handle)
+        run_id = _run_id(handle)
+        a_id = _default_marker_id(run_id, "aaa")
+
+        control = _activity_event(events, "control")
+        control_payload = (
+            control.activity_task_scheduled_event_attributes.input.payloads[0]
+        )
+        assert control_payload.metadata["encoding"] == b"custom"
+        assert control_payload.data == b"custom-converter-control"
+
+        activity_a = _activity_event(events, "activity-a")
+        activity_b = _activity_event(events, "activity-b")
+
+        # EG-LABEL-PAYLOAD-01: labels still go through the SDK default converter
+        assert _label_payload_of(activity_a, a_id) == ("json/plain", '"aaa"')
+        assert _label_payload_of(activity_b, "b-id") == ("json/plain", '"bbb"')
+
+
+async def test_label_payload_is_codec_encoded_but_ids_are_not(
+    env: WorkflowEnvironment,
+):
+    _require_event_groups_server(env)
+
+    codec = _WrappingPayloadCodec()
+    codec_client = await env.connect_client(
+        data_converter=DataConverter(payload_codec=codec)
+    )
+    async with new_worker(
+        codec_client,
+        LabelPayloadWorkflow,
+        activities=[noop_activity, control_activity],
+    ) as worker:
+        handle = await codec_client.start_workflow(
+            LabelPayloadWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle.result()
+        events = await _fetch_events(handle)
+        run_id = _run_id(handle)
+        a_id = _default_marker_id(run_id, "aaa")
+
+        activity_a = _activity_event(events, "activity-a")
+        activity_b = _activity_event(events, "activity-b")
+
+        # EG-LABEL-PAYLOAD-21: IDs are not codec-encoded
+        _assert_marker_ids(activity_a, _label_marker_id(a_id))
+        _assert_marker_ids(activity_b, _label_marker_id("b-id"))
+
+        # EG-LABEL-PAYLOAD-20: label payloads are processed by payload codecs
+        assert _label_payload_of(activity_a, a_id)[0] == "binary/wrapped"
+        assert _label_payload_of(activity_b, "b-id")[0] == "binary/wrapped"
+        decoded_a = (await codec.decode([_raw_label_payload(activity_a, a_id)]))[0]
+        decoded_b = (await codec.decode([_raw_label_payload(activity_b, "b-id")]))[0]
+        assert PayloadConverter.default.from_payload(decoded_a) == "aaa"
+        assert PayloadConverter.default.from_payload(decoded_b) == "bbb"
+
+
+####################################################################################################
+# 3. Explicit Event Group Scopes (`EG-SCOPE`)
+####################################################################################################
+
+
+@workflow.defn
+class ScopeBaselineWorkflow:
+    @workflow.run
+    async def run(self) -> None:
+        a = workflow.create_event_group("aaa")
+        with a.scope():
+            await _activity("activity")
+            await workflow.sleep(0.001)
+            await workflow.start_child_workflow(
+                NoopChildWorkflow.run,
+                id=f"{workflow.info().workflow_id}_child",
+            )
+
+
+async def test_commands_in_a_scope_carry_its_marker(
+    client: Client, env: WorkflowEnvironment
+):
+    _require_event_groups_server(env)
+
+    async with new_worker(
+        client,
+        ScopeBaselineWorkflow,
+        NoopChildWorkflow,
+        activities=[noop_activity],
+    ) as worker:
+        handle = await client.start_workflow(
+            ScopeBaselineWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle.result()
+        events = await _fetch_events(handle)
+        a = _label_marker(_default_marker_id(_run_id(handle), "aaa"), "aaa")
+
+        # EG-SCOPE-00: baseline only; per-command coverage lives in EG-COMMANDS
+        _assert_markers(_activity_event(events, "activity"), a)
+        _assert_markers(_single_event(events, EventType.EVENT_TYPE_TIMER_STARTED), a)
+        _assert_markers(
+            _single_event(
+                events, EventType.EVENT_TYPE_START_CHILD_WORKFLOW_EXECUTION_INITIATED
+            ),
+            a,
+        )
+
+
+@workflow.defn
+class NestedScopesWorkflow:
+    @workflow.run
+    async def run(self) -> None:
+        a = workflow.create_event_group("aaa")
+        b = workflow.create_event_group("bbb")
+        with a.scope():
+            await _activity("a-before")
+            with b.scope():
+                await _activity("a-b")
+            await _activity("a-after")
+        await _activity("outside")
+
+
+async def test_nesting_scopes_composes(client: Client, env: WorkflowEnvironment):
+    _require_event_groups_server(env)
+
+    async with new_worker(
+        client, NestedScopesWorkflow, activities=[noop_activity]
+    ) as worker:
+        handle = await client.start_workflow(
+            NestedScopesWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle.result()
+        events = await _fetch_events(handle)
+        run_id = _run_id(handle)
+        a = _label_marker(_default_marker_id(run_id, "aaa"), "aaa")
+        b = _label_marker(_default_marker_id(run_id, "bbb"), "bbb")
+
+        assert (
+            len(_events_of_type(events, EventType.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED))
+            == 4
+        )
+
+        # EG-SCOPE-01
+        _assert_markers(_activity_event(events, "a-before"), a)
+        _assert_markers(_activity_event(events, "a-b"), a, b)
+        _assert_markers(_activity_event(events, "a-after"), a)
+        _assert_markers(_activity_event(events, "outside"))
+
+
+@workflow.defn
+class ReenteredScopeWorkflow:
+    @workflow.run
+    async def run(self) -> None:
+        a = workflow.create_event_group("aaa")
+        with a.scope():
+            await _activity("a-before")
+            with a.scope():
+                await _activity("a-inner")
+            await _activity("a-after")
+        await _activity("outside")
+
+
+async def test_reentering_a_group_nests_correctly(
+    client: Client, env: WorkflowEnvironment
+):
+    _require_event_groups_server(env)
+
+    async with new_worker(
+        client, ReenteredScopeWorkflow, activities=[noop_activity]
+    ) as worker:
+        handle = await client.start_workflow(
+            ReenteredScopeWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle.result()
+        events = await _fetch_events(handle)
+        a = _label_marker(_default_marker_id(_run_id(handle), "aaa"), "aaa")
+
+        assert (
+            len(_events_of_type(events, EventType.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED))
+            == 4
+        )
+
+        # EG-SCOPE-02: inner re-entry still serializes the marker once
+        _assert_markers(_activity_event(events, "a-before"), a)
+        _assert_markers(_activity_event(events, "a-inner"), a)
+        _assert_markers(_activity_event(events, "a-after"), a)
+        _assert_markers(_activity_event(events, "outside"))
+
+
+@workflow.defn
+class ConcurrentScopesWorkflow:
+    @workflow.run
+    async def run(self) -> None:
+        a = workflow.create_event_group("aaa")
+        b = workflow.create_event_group("bbb")
+        c = workflow.create_event_group("ccc")
+        d = workflow.create_event_group("ddd")
+        e = workflow.create_event_group("eee")
+
+        async def left() -> None:
+            with b.scope():
+                with a.scope():
+                    with c.scope():
+                        await _activity("b-a-c")
+                    await _activity("b-a")
+                await _activity("b-after-a")
+
+        async def right() -> None:
+            with d.scope():
+                with a.scope():
+                    with e.scope():
+                        await _activity("d-a-e")
+                    await _activity("d-a")
+                await _activity("d-after-a")
+
+        await asyncio.gather(left(), right())
+        await _activity("outside")
+
+
+async def test_a_group_can_be_scoped_from_two_concurrent_branches(
+    client: Client, env: WorkflowEnvironment
+):
+    _require_event_groups_server(env)
+
+    async with new_worker(
+        client, ConcurrentScopesWorkflow, activities=[noop_activity]
+    ) as worker:
+        handle = await client.start_workflow(
+            ConcurrentScopesWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle.result()
+        events = await _fetch_events(handle)
+        run_id = _run_id(handle)
+        a = _label_marker(_default_marker_id(run_id, "aaa"), "aaa")
+        b = _label_marker(_default_marker_id(run_id, "bbb"), "bbb")
+        c = _label_marker(_default_marker_id(run_id, "ccc"), "ccc")
+        d = _label_marker(_default_marker_id(run_id, "ddd"), "ddd")
+        e = _label_marker(_default_marker_id(run_id, "eee"), "eee")
+
+        assert (
+            len(_events_of_type(events, EventType.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED))
+            == 7
+        )
+
+        # EG-SCOPE-03: a mutable "currently active groups" stack would cross-contaminate here
+        _assert_markers(_activity_event(events, "b-a-c"), b, a, c)
+        _assert_markers(_activity_event(events, "b-a"), b, a)
+        _assert_markers(_activity_event(events, "b-after-a"), b)
+        _assert_markers(_activity_event(events, "d-a-e"), d, a, e)
+        _assert_markers(_activity_event(events, "d-a"), d, a)
+        _assert_markers(_activity_event(events, "d-after-a"), d)
+        _assert_markers(_activity_event(events, "outside"))
+
+
+@workflow.defn
+class DetachedTaskScopeWorkflow:
+    @workflow.run
+    async def run(self) -> None:
+        a = workflow.create_event_group("aaa")
+        released = False
+        started = False
+
+        async def task() -> None:
+            nonlocal started
+            await _activity("inside-before")
+            started = True
+            await workflow.wait_condition(lambda: released)
+            await _activity("inside-after")
+
+        with a.scope():
+            running = asyncio.create_task(task())
+            await workflow.wait_condition(lambda: started)
+        released = True
+        await running
+        await _activity("outside")
+
+
+async def test_a_task_started_inside_a_scope_keeps_it_after_exit(
+    client: Client, env: WorkflowEnvironment
+):
+    _require_event_groups_server(env)
+
+    async with new_worker(
+        client, DetachedTaskScopeWorkflow, activities=[noop_activity]
+    ) as worker:
+        handle = await client.start_workflow(
+            DetachedTaskScopeWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle.result()
+        events = await _fetch_events(handle)
+        a = _label_marker(_default_marker_id(_run_id(handle), "aaa"), "aaa")
+
+        assert (
+            len(_events_of_type(events, EventType.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED))
+            == 3
+        )
+
+        # EG-SCOPE-04: membership is captured when the task is started
+        _assert_markers(_activity_event(events, "inside-before"), a)
+        _assert_markers(_activity_event(events, "inside-after"), a)
+        _assert_markers(_activity_event(events, "outside"))
+
+
+@workflow.defn
+class OutsiderTaskScopeWorkflow:
+    @workflow.run
+    async def run(self) -> None:
+        a = workflow.create_event_group("aaa")
+        release = False
+
+        async def outsider() -> None:
+            await workflow.wait_condition(lambda: release)
+            await _activity("outside-task")
+
+        running = asyncio.create_task(outsider())
+        with a.scope():
+            await _activity("in-a")
+            release = True
+            await running
+
+
+async def test_a_task_created_outside_a_scope_does_not_inherit_it(
+    client: Client, env: WorkflowEnvironment
+):
+    _require_event_groups_server(env)
+
+    async with new_worker(
+        client, OutsiderTaskScopeWorkflow, activities=[noop_activity]
+    ) as worker:
+        handle = await client.start_workflow(
+            OutsiderTaskScopeWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle.result()
+        events = await _fetch_events(handle)
+        a = _label_marker(_default_marker_id(_run_id(handle), "aaa"), "aaa")
+
+        assert (
+            len(_events_of_type(events, EventType.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED))
+            == 2
+        )
+
+        # EG-SCOPE-05: membership follows the context the code was started in
+        _assert_markers(_activity_event(events, "in-a"), a)
+        _assert_markers(_activity_event(events, "outside-task"))
+
+
+@workflow.defn
+class ThrowingScopeWorkflow:
+    @workflow.run
+    async def run(self) -> None:
+        a = workflow.create_event_group("aaa")
+        b = workflow.create_event_group("bbb")
+        with a.scope():
+            try:
+                with b.scope():
+                    await _activity("a-b")
+                    raise RuntimeError("boom")
+            except RuntimeError:
+                pass
+            await _activity("a-after")
+
+
+async def test_a_scope_unwinds_cleanly_when_its_body_throws(
+    client: Client, env: WorkflowEnvironment
+):
+    _require_event_groups_server(env)
+
+    async with new_worker(
+        client, ThrowingScopeWorkflow, activities=[noop_activity]
+    ) as worker:
+        handle = await client.start_workflow(
+            ThrowingScopeWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle.result()
+        events = await _fetch_events(handle)
+        run_id = _run_id(handle)
+        a = _label_marker(_default_marker_id(run_id, "aaa"), "aaa")
+        b = _label_marker(_default_marker_id(run_id, "bbb"), "bbb")
+
+        assert (
+            len(_events_of_type(events, EventType.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED))
+            == 2
+        )
+
+        # EG-SCOPE-06
+        _assert_markers(_activity_event(events, "a-b"), a, b)
+        _assert_markers(_activity_event(events, "a-after"), a)
+
+
+####################################################################################################
+# 4. Implicit Event Groups (`EG-IMPLICIT`)
+####################################################################################################
+
+
+@workflow.defn
+class StaticSignalHandlerWorkflow:
+    def __init__(self) -> None:
+        self._done = False
+
+    @workflow.run
+    async def run(self) -> None:
+        await _activity("from-main-before-signal")
+        await workflow.wait_condition(lambda: self._done)
+        await _activity("from-main-after-signal")
+
+    @workflow.signal
+    async def my_signal(self) -> None:
+        await _activity("from-static-signal")
+        a = workflow.create_event_group("aaa")
+        with a.scope():
+            await _activity("from-static-signal-scoped")
+        self._done = True
+
+
+async def test_static_signal_handler_implicit_group(
+    client: Client, env: WorkflowEnvironment
+):
+    _require_event_groups_server(env)
+
+    async with new_worker(
+        client, StaticSignalHandlerWorkflow, activities=[noop_activity]
+    ) as worker:
+        handle = await client.start_workflow(
+            StaticSignalHandlerWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle.signal(StaticSignalHandlerWorkflow.my_signal)
+        await handle.result()
+        events = await _fetch_events(handle)
+        signal = _event_marker(_signaled_event_ids(events)[0])
+        a = _label_marker(_default_marker_id(_run_id(handle), "aaa"), "aaa")
+
+        # EG-IMPLICIT-00
+        _assert_markers(_activity_event(events, "from-static-signal"), signal)
+        # EG-IMPLICIT-01
+        _assert_markers(_activity_event(events, "from-static-signal-scoped"), signal, a)
+        # EG-IMPLICIT-30
+        _assert_markers(_activity_event(events, "from-main-before-signal"))
+        _assert_markers(_activity_event(events, "from-main-after-signal"))
+
+
+@workflow.defn
+class RuntimeSignalHandlerWorkflow:
+    def __init__(self) -> None:
+        self._done = False
+
+    @workflow.run
+    async def run(self) -> None:
+        outside = workflow.create_event_group("outside")
+        inside = workflow.create_event_group("inside")
+
+        async def on_signal() -> None:
+            await _activity("from-runtime-signal")
+            with inside.scope():
+                await _activity("from-runtime-signal-scoped")
+            self._done = True
+
+        with outside.scope():
+            workflow.set_signal_handler("mySignal", on_signal)
+            await _activity("in-outside")
+
+        await _activity("from-main-before-signal")
+        await workflow.wait_condition(lambda: self._done)
+        await _activity("from-main-after-signal")
+
+
+async def test_runtime_signal_handler_implicit_group(
+    client: Client, env: WorkflowEnvironment
+):
+    _require_event_groups_server(env)
+
+    async with new_worker(
+        client, RuntimeSignalHandlerWorkflow, activities=[noop_activity]
+    ) as worker:
+        handle = await client.start_workflow(
+            RuntimeSignalHandlerWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle.signal("mySignal")
+        await handle.result()
+        events = await _fetch_events(handle)
+        run_id = _run_id(handle)
+        signal_ids = _signaled_event_ids(events)
+        assert len(signal_ids) == 1
+        signal = _event_marker(signal_ids[0])
+        outside = _label_marker(_default_marker_id(run_id, "outside"), "outside")
+        inside = _label_marker(_default_marker_id(run_id, "inside"), "inside")
+
+        # EG-IMPLICIT-10: handler carries the signaled event, not the registration scope
+        _assert_markers(_activity_event(events, "from-runtime-signal"), signal)
+        _assert_markers(_activity_event(events, "in-outside"), outside)
+        # EG-IMPLICIT-11
+        _assert_markers(
+            _activity_event(events, "from-runtime-signal-scoped"), signal, inside
+        )
+        # EG-IMPLICIT-30
+        _assert_markers(_activity_event(events, "from-main-before-signal"))
+        _assert_markers(_activity_event(events, "from-main-after-signal"))
+
+
+@workflow.defn
+class BufferedSignalWorkflow:
+    def __init__(self) -> None:
+        self._unblocked = False
+        self._handled = False
+
+    @workflow.run
+    async def run(self) -> None:
+        workflow.set_signal_handler("unblock", self._unblock)
+        await workflow.wait_condition(lambda: self._unblocked)
+
+        async def on_signal() -> None:
+            await _activity("from-runtime-signal")
+            self._handled = True
+
+        workflow.set_signal_handler("mySignal", on_signal)
+        await workflow.wait_condition(lambda: self._handled)
+
+    def _unblock(self) -> None:
+        self._unblocked = True
+
+
+async def test_buffered_signal_keeps_its_original_implicit_marker(
+    client: Client, env: WorkflowEnvironment
+):
+    _require_event_groups_server(env)
+
+    async with new_worker(
+        client, BufferedSignalWorkflow, activities=[noop_activity]
+    ) as worker:
+        handle = await client.start_workflow(
+            BufferedSignalWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle.signal("mySignal")
+        await handle.signal("unblock")
+        await handle.result()
+        events = await _fetch_events(handle)
+        signal_ids = _signaled_event_ids(events)
+        assert len(signal_ids) == 2
+        # mySignal is sent first, so it is the first signaled event
+        signal = _event_marker(signal_ids[0])
+
+        # EG-IMPLICIT-12
+        _assert_markers(_activity_event(events, "from-runtime-signal"), signal)
+
+
+@workflow.defn
+class CatchAllSignalWorkflow:
+    def __init__(self) -> None:
+        self._done = False
+
+    @workflow.run
+    async def run(self) -> None:
+        await workflow.wait_condition(lambda: self._done)
+
+    @workflow.signal(dynamic=True)
+    async def on_any_signal(self, _name: str, _args: Sequence[RawValue]) -> None:
+        await _activity("from-catch-all-signal")
+        self._done = True
+
+
+async def test_catch_all_signal_handler_implicit_group(
+    client: Client, env: WorkflowEnvironment
+):
+    _require_event_groups_server(env)
+
+    async with new_worker(
+        client, CatchAllSignalWorkflow, activities=[noop_activity]
+    ) as worker:
+        handle = await client.start_workflow(
+            CatchAllSignalWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle.signal("non-existent-signal")
+        await handle.result()
+        events = await _fetch_events(handle)
+        signal = _event_marker(_signaled_event_ids(events)[0])
+
+        # EG-IMPLICIT-20
+        _assert_markers(_activity_event(events, "from-catch-all-signal"), signal)
+
+
+@workflow.defn
+class StaticUpdateHandlerWorkflow:
+    def __init__(self) -> None:
+        self._done = False
+
+    @workflow.run
+    async def run(self) -> None:
+        await _activity("from-main-before-update")
+        await workflow.wait_condition(lambda: self._done)
+        await _activity("from-main-after-update")
+
+    @workflow.update
+    async def my_update(self) -> None:
+        await _activity("from-static-update")
+        inside = workflow.create_event_group("inside")
+        with inside.scope():
+            await _activity("from-static-update-scoped")
+        self._done = True
+
+
+async def test_static_update_handler_implicit_group(
+    client: Client, env: WorkflowEnvironment
+):
+    _require_event_groups_server(env)
+
+    update_id = "static-update-1"
+    async with new_worker(
+        client, StaticUpdateHandlerWorkflow, activities=[noop_activity]
+    ) as worker:
+        handle = await client.start_workflow(
+            StaticUpdateHandlerWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle.execute_update(StaticUpdateHandlerWorkflow.my_update, id=update_id)
+        await handle.result()
+        events = await _fetch_events(handle)
+        update = _update_marker(update_id)
+        inside = _label_marker(_default_marker_id(_run_id(handle), "inside"), "inside")
+
+        # EG-IMPLICIT-50
+        _assert_markers(_activity_event(events, "from-static-update"), update)
+        # EG-IMPLICIT-51
+        _assert_markers(
+            _activity_event(events, "from-static-update-scoped"), update, inside
+        )
+        # EG-IMPLICIT-80
+        _assert_markers(_activity_event(events, "from-main-before-update"))
+        _assert_markers(_activity_event(events, "from-main-after-update"))
+
+
+@workflow.defn
+class RuntimeUpdateHandlerWorkflow:
+    def __init__(self) -> None:
+        self._done = False
+
+    @workflow.run
+    async def run(self) -> None:
+        outside = workflow.create_event_group("outside")
+        inside = workflow.create_event_group("inside")
+
+        async def on_update() -> None:
+            await _activity("from-runtime-update")
+            with inside.scope():
+                await _activity("from-runtime-update-scoped")
+            self._done = True
+
+        with outside.scope():
+            workflow.set_update_handler("myUpdate", on_update)
+            await _activity("in-outside")
+
+        await _activity("from-main-before-update")
+        await workflow.wait_condition(lambda: self._done)
+        await _activity("from-main-after-update")
+
+
+async def test_runtime_update_handler_implicit_group(
+    client: Client, env: WorkflowEnvironment
+):
+    _require_event_groups_server(env)
+
+    update_id = "runtime-update-1"
+    async with new_worker(
+        client, RuntimeUpdateHandlerWorkflow, activities=[noop_activity]
+    ) as worker:
+        handle = await client.start_workflow(
+            RuntimeUpdateHandlerWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+
+        # Updates that arrive before registration are rejected, not buffered.
+        async def handler_registered() -> None:
+            events = await _fetch_events(handle)
+            assert any(
+                e.event_type == EventType.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED
+                and e.activity_task_scheduled_event_attributes.activity_id
+                == "in-outside"
+                for e in events
+            )
+
+        await assert_eventually(handler_registered)
+        await handle.execute_update("myUpdate", id=update_id)
+        await handle.result()
+        events = await _fetch_events(handle)
+        run_id = _run_id(handle)
+        update = _update_marker(update_id)
+        outside = _label_marker(_default_marker_id(run_id, "outside"), "outside")
+        inside = _label_marker(_default_marker_id(run_id, "inside"), "inside")
+
+        # EG-IMPLICIT-60
+        _assert_markers(_activity_event(events, "from-runtime-update"), update)
+        _assert_markers(_activity_event(events, "in-outside"), outside)
+        # EG-IMPLICIT-61
+        _assert_markers(
+            _activity_event(events, "from-runtime-update-scoped"), update, inside
+        )
+        # EG-IMPLICIT-80
+        _assert_markers(_activity_event(events, "from-main-before-update"))
+        _assert_markers(_activity_event(events, "from-main-after-update"))
+
+
+@workflow.defn
+class CatchAllUpdateWorkflow:
+    def __init__(self) -> None:
+        self._done = False
+
+    @workflow.run
+    async def run(self) -> None:
+        await workflow.wait_condition(lambda: self._done)
+
+    @workflow.update(dynamic=True)
+    async def on_any_update(self, _name: str, _args: Sequence[RawValue]) -> None:
+        await _activity("from-catch-all-update")
+        self._done = True
+
+
+async def test_catch_all_update_handler_implicit_group(
+    client: Client, env: WorkflowEnvironment
+):
+    _require_event_groups_server(env)
+
+    update_id = "catch-all-update-1"
+    async with new_worker(
+        client, CatchAllUpdateWorkflow, activities=[noop_activity]
+    ) as worker:
+        handle = await client.start_workflow(
+            CatchAllUpdateWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle.execute_update("non-existent-update", id=update_id)
+        await handle.result()
+        events = await _fetch_events(handle)
+
+        # EG-IMPLICIT-70
+        _assert_markers(
+            _activity_event(events, "from-catch-all-update"), _update_marker(update_id)
+        )
+
+
+####################################################################################################
+# 5. Event Group Marker Aggregation (`EG-AGGREGATION`)
+####################################################################################################
+
+
+@workflow.defn
+class AggregationWorkflow:
+    @workflow.run
+    async def run(self) -> None:
+        a1 = workflow.create_event_group("aaa")
+        a2 = workflow.create_event_group("aaa")
+        b1 = workflow.create_event_group("bbb1", id="b-id")
+        b2 = workflow.create_event_group("bbb2", id="b-id")
+
+        await _activity("direct-duplicates", [a2, b1, a1, b1, a2, a1])
+
+        with a1.scope():
+            with a2.scope():
+                with b1.scope():
+                    await _activity("nested-scopes")
+
+        with a1.scope():
+            with b1.scope():
+                await _activity("scope-and-direct-b", [b1])
+                await _activity("scope-and-direct-a-b", [b1, a1])
+
+        await _activity("same-instance-twice", [a1, a1])
+        await _activity("same-id-direct", [b1, b2])
+        with b1.scope():
+            await _activity("same-id-scope-and-direct", [b2])
+
+
+async def test_markers_dedupe_by_id(client: Client, env: WorkflowEnvironment):
+    _require_event_groups_server(env)
+
+    async with new_worker(
+        client, AggregationWorkflow, activities=[noop_activity]
+    ) as worker:
+        handle = await client.start_workflow(
+            AggregationWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle.result()
+        events = await _fetch_events(handle)
+        a = _label_marker(_default_marker_id(_run_id(handle), "aaa"), "aaa")
+        b = _label_marker("b-id", "bbb1")
+        both = (a, b)
+
+        # EG-AGGREGATION-00
+        _assert_markers(_activity_event(events, "direct-duplicates"), *both)
+        # EG-AGGREGATION-01
+        _assert_markers(_activity_event(events, "nested-scopes"), *both)
+        # EG-AGGREGATION-02
+        _assert_markers(_activity_event(events, "scope-and-direct-b"), *both)
+        _assert_markers(_activity_event(events, "scope-and-direct-a-b"), *both)
+        # EG-AGGREGATION-03
+        _assert_markers(_activity_event(events, "same-instance-twice"), a)
+        # EG-AGGREGATION-04: compare IDs only; which label is emitted is unspecified
+        _assert_marker_ids(
+            _activity_event(events, "same-id-direct"), _label_marker_id("b-id")
+        )
+        _assert_marker_ids(
+            _activity_event(events, "same-id-scope-and-direct"),
+            _label_marker_id("b-id"),
+        )
+
+
+####################################################################################################
+# 6. Command Type Coverage (`EG-COMMANDS`)
+#
+# EG-COMMANDS-23 and EG-COMMANDS-24 do not apply: Core-based SDKs have no version/sideEffect API.
+# Python continue_as_new always takes options, so there is no short-form counterpart of EG-COMMANDS-40.
+# ExternalWorkflowHandle.signal/cancel do not take event_groups; EG-COMMANDS-06/07 assert ambient only.
+####################################################################################################
+
+
+@workflow.defn
+class TimerCommandsWorkflow:
+    @workflow.run
+    async def run(self) -> None:
+        direct = workflow.create_event_group("direct")
+        scope = workflow.create_event_group("scope")
+        with scope.scope():
+            await workflow.sleep(0.001, event_groups=[direct])
+            try:
+                await workflow.wait_condition(
+                    lambda: False, timeout=0.001, event_groups=[direct]
+                )
+            except asyncio.TimeoutError:
+                pass
+            # A 1ms sleep is the timeout; cancelling the 60s task is the cancel
+            # command. Avoid asyncio.wait_for so the timeout timer is a normal
+            # sleep and only carries the ambient scope.
+            long = asyncio.create_task(workflow.sleep(60, event_groups=[direct]))
+            await workflow.sleep(0.001)
+            long.cancel()
+            await _swallow(long)
+
+
+async def test_timer_commands_carry_markers(client: Client, env: WorkflowEnvironment):
+    _require_event_groups_server(env)
+
+    async with new_worker(client, TimerCommandsWorkflow) as worker:
+        handle = await client.start_workflow(
+            TimerCommandsWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle.result()
+        events = await _fetch_events(handle)
+        run_id = _run_id(handle)
+        both = (
+            _label_marker(_default_marker_id(run_id, "direct"), "direct"),
+            _label_marker(_default_marker_id(run_id, "scope"), "scope"),
+        )
+        ambient = (_label_marker(_default_marker_id(run_id, "scope"), "scope"),)
+
+        timers = _events_of_type(events, EventType.EVENT_TYPE_TIMER_STARTED)
+        # sleep, wait_condition timeout, wait_for's 1ms timeout, the cancelled 60s sleep
+        assert len(timers) == 4
+        cancels = _events_of_type(events, EventType.EVENT_TYPE_TIMER_CANCELED)
+        assert len(cancels) == 1
+
+        # EG-COMMANDS-00 and EG-COMMANDS-01 run sequentially, so they are the first two timers
+        _assert_markers(timers[0], *both)
+        _assert_markers(timers[1], *both)
+        # EG-COMMANDS-00-CANCEL: wait_for starts both remaining timers in one task, so select by set
+        rest = timers[2:]
+        ambient_timers = [t for t in rest if _markers(t) == sorted(ambient)]
+        both_timers = [t for t in rest if _markers(t) == sorted(both)]
+        assert len(ambient_timers) == 1
+        assert len(both_timers) == 1
+        _assert_markers(cancels[0], *both)
+
+
+@workflow.defn
+class ActivityCommandsWorkflow:
+    @workflow.run
+    async def run(self) -> None:
+        direct = workflow.create_event_group("direct")
+        scope = workflow.create_event_group("scope")
+        with scope.scope():
+            await workflow.execute_activity(
+                noop_activity,
+                start_to_close_timeout=_ACT_TIMEOUT,
+                schedule_to_start_timeout=timedelta(seconds=10),
+                event_groups=[direct],
+                activity_id="activity",
+            )
+            await _swallow(
+                asyncio.wait_for(
+                    workflow.execute_activity(
+                        sleep_activity,
+                        start_to_close_timeout=_ACT_TIMEOUT,
+                        schedule_to_start_timeout=timedelta(seconds=10),
+                        cancellation_type=workflow.ActivityCancellationType.TRY_CANCEL,
+                        event_groups=[direct],
+                        activity_id="activity-cancelled-sleep-5s",
+                    ),
+                    timeout=0.001,
+                )
+            )
+
+
+async def test_activity_commands_carry_markers(
+    client: Client, env: WorkflowEnvironment
+):
+    _require_event_groups_server(env)
+
+    async with new_worker(
+        client,
+        ActivityCommandsWorkflow,
+        activities=[noop_activity, sleep_activity],
+    ) as worker:
+        handle = await client.start_workflow(
+            ActivityCommandsWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle.result()
+        events = await _fetch_events(handle)
+        run_id = _run_id(handle)
+        both = (
+            _label_marker(_default_marker_id(run_id, "direct"), "direct"),
+            _label_marker(_default_marker_id(run_id, "scope"), "scope"),
+        )
+
+        # EG-COMMANDS-02
+        _assert_markers(_activity_event(events, "activity"), *both)
+        # EG-COMMANDS-02-CANCEL
+        _assert_markers(_activity_event(events, "activity-cancelled-sleep-5s"), *both)
+        _assert_markers(
+            _single_event(events, EventType.EVENT_TYPE_ACTIVITY_TASK_CANCEL_REQUESTED),
+            *both,
+        )
+
+
+@workflow.defn
+class LocalActivityCommandsWorkflow:
+    @workflow.run
+    async def run(self) -> None:
+        direct = workflow.create_event_group("direct")
+        scope = workflow.create_event_group("scope")
+        with scope.scope():
+            await workflow.execute_local_activity(
+                noop_activity,
+                start_to_close_timeout=_ACT_TIMEOUT,
+                event_groups=[direct],
+                activity_id="local-activity",
+            )
+
+            cancel_trigger = workflow.create_event_group("cancel-trigger")
+            cancelled_la = workflow.create_event_group("cancelled-la")
+            sleeping = asyncio.create_task(
+                workflow.execute_local_activity(
+                    sleep_activity,
+                    start_to_close_timeout=_ACT_TIMEOUT,
+                    cancellation_type=workflow.ActivityCancellationType.TRY_CANCEL,
+                    event_groups=[direct, cancelled_la],
+                    activity_id="cancelled-local-activity-sleep-5s",
+                )
+            )
+
+            async def trigger() -> None:
+                await workflow.execute_local_activity(
+                    noop_activity,
+                    start_to_close_timeout=_ACT_TIMEOUT,
+                    event_groups=[direct, cancel_trigger],
+                    activity_id="cancel-trigger",
+                )
+                sleeping.cancel()
+
+            await asyncio.gather(trigger(), _swallow(sleeping))
+
+            await workflow.execute_local_activity(
+                fail_first_activity,
+                start_to_close_timeout=_ACT_TIMEOUT,
+                local_retry_threshold=timedelta(milliseconds=1),
+                retry_policy=RetryPolicy(
+                    initial_interval=timedelta(seconds=1),
+                    backoff_coefficient=1,
+                    maximum_attempts=2,
+                ),
+                event_groups=[direct],
+                activity_id="backoff-local-activity-fail-first-attempt",
+            )
+
+
+async def test_local_activity_commands_carry_markers(
+    client: Client, env: WorkflowEnvironment
+):
+    _require_event_groups_server(env)
+
+    async with new_worker(
+        client,
+        LocalActivityCommandsWorkflow,
+        activities=[noop_activity, sleep_activity, fail_first_activity],
+    ) as worker:
+        handle = await client.start_workflow(
+            LocalActivityCommandsWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+            task_timeout=timedelta(seconds=5),
+        )
+        await handle.result()
+        events = await _fetch_events(handle)
+        run_id = _run_id(handle)
+        both = (
+            _label_marker(_default_marker_id(run_id, "direct"), "direct"),
+            _label_marker(_default_marker_id(run_id, "scope"), "scope"),
+        )
+        cancel_trigger = (
+            *both,
+            _label_marker(
+                _default_marker_id(run_id, "cancel-trigger"), "cancel-trigger"
+            ),
+        )
+        cancelled_la = (
+            *both,
+            _label_marker(_default_marker_id(run_id, "cancelled-la"), "cancelled-la"),
+        )
+
+        local_acts = _markers_named(events, "core_local_activity")
+        # plain complete, cancel-trigger complete, cancelled LA, backoff fail, backoff success
+        assert len(local_acts) == 5
+
+        # EG-COMMANDS-03
+        _assert_markers(local_acts[0], *both)
+        # EG-COMMANDS-03-CANCEL
+        _assert_markers(local_acts[1], *cancel_trigger)
+        _assert_markers(local_acts[2], *cancelled_la)
+        # EG-COMMANDS-03-BACKOFF: plan's 10s interval vs 5s WFT timeout is the same Core branch;
+        # local_retry_threshold of 1ms reaches it without waiting 10s.
+        backoff_timer = _events_of_type(events, EventType.EVENT_TYPE_TIMER_STARTED)
+        assert len(backoff_timer) == 1
+        _assert_markers(backoff_timer[0], *both)
+        _assert_markers(local_acts[3], *both)
+        _assert_markers(local_acts[4], *both)
+
+
+@workflow.defn
+class ChildWorkflowCommandsWorkflow:
+    @workflow.run
+    async def run(self) -> None:
+        direct = workflow.create_event_group("direct")
+        scope = workflow.create_event_group("scope")
+        with scope.scope():
+            await workflow.start_child_workflow(
+                NoopChildWorkflow.run,
+                id=f"{workflow.info().workflow_id}_child",
+                event_groups=[direct],
+            )
+            await _swallow(
+                asyncio.wait_for(
+                    workflow.execute_child_workflow(
+                        SleepChildWorkflow.run,
+                        id=f"{workflow.info().workflow_id}_child_cancel",
+                        cancellation_type=workflow.ChildWorkflowCancellationType.WAIT_CANCELLATION_REQUESTED,
+                        event_groups=[direct],
+                    ),
+                    timeout=0.001,
+                )
+            )
+
+
+async def test_child_workflow_commands_carry_markers(
+    client: Client, env: WorkflowEnvironment
+):
+    _require_event_groups_server(env)
+
+    async with new_worker(
+        client,
+        ChildWorkflowCommandsWorkflow,
+        NoopChildWorkflow,
+        SleepChildWorkflow,
+    ) as worker:
+        handle = await client.start_workflow(
+            ChildWorkflowCommandsWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle.result()
+        events = await _fetch_events(handle)
+        run_id = _run_id(handle)
+        both = (
+            _label_marker(_default_marker_id(run_id, "direct"), "direct"),
+            _label_marker(_default_marker_id(run_id, "scope"), "scope"),
+        )
+
+        initiated = _events_of_type(
+            events, EventType.EVENT_TYPE_START_CHILD_WORKFLOW_EXECUTION_INITIATED
+        )
+        assert len(initiated) == 2
+        # EG-COMMANDS-04
+        _assert_markers(initiated[0], *both)
+        # EG-COMMANDS-04-CANCEL
+        _assert_markers(initiated[1], *both)
+        _assert_markers(
+            _single_event(
+                events,
+                EventType.EVENT_TYPE_REQUEST_CANCEL_EXTERNAL_WORKFLOW_EXECUTION_INITIATED,
+            ),
+            *both,
+        )
+
+
+@nexusrpc.handler.service_handler
+class EventGroupsNexusService:
+    @nexusrpc.handler.sync_operation
+    async def nexus_operation(
+        self, _ctx: nexusrpc.handler.StartOperationContext, _input: None
+    ) -> None:
+        return None
+
+    @nexusrpc.handler.sync_operation
+    async def nexus_operation_sleep_5s(
+        self, _ctx: nexusrpc.handler.StartOperationContext, _input: None
+    ) -> None:
+        await asyncio.sleep(5)
+
+
+@workflow.defn
+class NexusCommandsWorkflow:
+    @workflow.run
+    async def run(self, endpoint: str) -> None:
+        direct = workflow.create_event_group("direct")
+        scope = workflow.create_event_group("scope")
+        nexus_client = workflow.create_nexus_client(
+            service=EventGroupsNexusService, endpoint=endpoint
+        )
+        with scope.scope():
+            await nexus_client.execute_operation(
+                EventGroupsNexusService.nexus_operation,
+                None,
+                event_groups=[direct],
+            )
+            # Don't await start/execute to completion: a sync sleeper would finish
+            # before cancel, and Core drops a cancel issued in the same WFT as
+            # schedule. The 1ms sleep forces a WFT boundary after schedule.
+            running = asyncio.create_task(
+                nexus_client.execute_operation(
+                    EventGroupsNexusService.nexus_operation_sleep_5s,
+                    None,
+                    cancellation_type=workflow.NexusOperationCancellationType.TRY_CANCEL,
+                    event_groups=[direct],
+                )
+            )
+            await workflow.sleep(0.001)
+            running.cancel()
+            await _swallow(running)
+
+
+@pytest.mark.requires_local_server
+async def test_nexus_operation_commands_carry_markers(
+    client: Client, env: WorkflowEnvironment
+):
+    if env.supports_time_skipping:
+        pytest.skip("Nexus tests don't work with the time-skipping server")
+
+    task_queue = str(uuid.uuid4())
+    endpoint = make_nexus_endpoint_name(task_queue)
+    await env.create_nexus_endpoint(endpoint, task_queue)
+    async with Worker(
+        client,
+        task_queue=task_queue,
+        workflows=[NexusCommandsWorkflow],
+        nexus_service_handlers=[EventGroupsNexusService()],
+    ):
+        handle = await client.start_workflow(
+            NexusCommandsWorkflow.run,
+            endpoint,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=task_queue,
+        )
+        await handle.result()
+        events = await _fetch_events(handle)
+        run_id = _run_id(handle)
+        both = (
+            _label_marker(_default_marker_id(run_id, "direct"), "direct"),
+            _label_marker(_default_marker_id(run_id, "scope"), "scope"),
+        )
+
+        scheduled = _events_of_type(
+            events, EventType.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED
+        )
+        assert len(scheduled) == 2
+        # EG-COMMANDS-05
+        _assert_markers(scheduled[0], *both)
+        # EG-COMMANDS-05-CANCEL
+        _assert_markers(scheduled[1], *both)
+        _assert_markers(
+            _single_event(
+                events, EventType.EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUESTED
+            ),
+            *both,
+        )
+
+
+@workflow.defn
+class AmbientOnlyCommandsWorkflow:
+    @workflow.run
+    async def run(self) -> None:
+        scope = workflow.create_event_group("scope")
+        with scope.scope():
+            child = await workflow.start_child_workflow(
+                SleepChildWorkflow.run,
+                id=f"{workflow.info().workflow_id}_child",
+            )
+            await child.signal("noop")
+            # External handle cancel of a live child: missing-workflow not-found
+            # fails the WFT in a way that is not cleanly catchable here.
+            await workflow.get_external_workflow_handle(child.id).cancel()
+            workflow.upsert_memo({"some-key": "some-value"})
+            workflow.upsert_search_attributes(
+                [SearchAttributeKey.for_bool("CustomBoolField").value_set(False)]
+            )
+            workflow.patched("my-patch-1")
+            workflow.deprecate_patch("my-patch-2")
+
+
+async def test_apis_without_options_carry_ambient_markers(
+    client: Client, env: WorkflowEnvironment
+):
+    _require_event_groups_server(env)
+    await ensure_search_attributes_present(
+        client, SearchAttributeKey.for_bool("CustomBoolField")
+    )
+
+    async with new_worker(
+        client, AmbientOnlyCommandsWorkflow, SleepChildWorkflow
+    ) as worker:
+        handle = await client.start_workflow(
+            AmbientOnlyCommandsWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle.result()
+        events = await _fetch_events(handle)
+        ambient = (
+            _label_marker(_default_marker_id(_run_id(handle), "scope"), "scope"),
+        )
+
+        # EG-COMMANDS-06, EG-COMMANDS-07: no direct-attach option on the external handle
+        _assert_markers(
+            _single_event(
+                events,
+                EventType.EVENT_TYPE_SIGNAL_EXTERNAL_WORKFLOW_EXECUTION_INITIATED,
+            ),
+            *ambient,
+        )
+        _assert_markers(
+            _single_event(
+                events,
+                EventType.EVENT_TYPE_REQUEST_CANCEL_EXTERNAL_WORKFLOW_EXECUTION_INITIATED,
+            ),
+            *ambient,
+        )
+        # EG-COMMANDS-20
+        _assert_markers(
+            _single_event(events, EventType.EVENT_TYPE_WORKFLOW_PROPERTIES_MODIFIED),
+            *ambient,
+        )
+        # EG-COMMANDS-22
+        patches = _markers_named(events, "core_patch")
+        assert len(patches) == 2
+        for patch in patches:
+            _assert_markers(patch, *ambient)
+        # EG-COMMANDS-21 plus the two TemporalChangeVersion upserts beside the patches
+        upserts = _events_of_type(
+            events, EventType.EVENT_TYPE_UPSERT_WORKFLOW_SEARCH_ATTRIBUTES
+        )
+        assert len(upserts) == 3
+        for upsert in upserts:
+            _assert_markers(upsert, *ambient)
+
+
+@workflow.defn
+class ContinueAsNewCommandsWorkflow:
+    @workflow.run
+    async def run(self, second_run: bool = False) -> None:
+        if second_run:
+            return
+        direct = workflow.create_event_group("direct")
+        scope = workflow.create_event_group("scope")
+        with scope.scope():
+            workflow.continue_as_new(True, event_groups=[direct])
+
+
+async def test_continue_as_new_carries_markers(
+    client: Client, env: WorkflowEnvironment
+):
+    _require_event_groups_server(env)
+
+    async with new_worker(client, ContinueAsNewCommandsWorkflow) as worker:
+        handle = await client.start_workflow(
+            ContinueAsNewCommandsWorkflow.run,
+            False,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        await handle.result()
+        # ContinuedAsNew lives on the first run, which is also the run whose id the markers used
+        first_run = client.get_workflow_handle(handle.id, run_id=_run_id(handle))
+        events = await _fetch_events(first_run)
+        run_id = _run_id(handle)
+        both = (
+            _label_marker(_default_marker_id(run_id, "direct"), "direct"),
+            _label_marker(_default_marker_id(run_id, "scope"), "scope"),
+        )
+
+        # EG-COMMANDS-40
+        _assert_markers(
+            _single_event(
+                events, EventType.EVENT_TYPE_WORKFLOW_EXECUTION_CONTINUED_AS_NEW
+            ),
+            *both,
+        )
+
+
+####################################################################################################
+# Language-specific
+####################################################################################################
+
+
+@workflow.defn
+class EmptyLabelWorkflow:
+    @workflow.run
+    async def run(self) -> str:
+        try:
+            workflow.create_event_group("")
+        except ValueError as err:
+            return str(err)
+        return "no error"
+
+
+async def test_event_group_rejects_empty_label(
+    client: Client, env: WorkflowEnvironment
+):
+    _require_event_groups_server(env)
+
+    async with new_worker(client, EmptyLabelWorkflow) as worker:
+        assert "Event group label cannot be empty" == await client.execute_workflow(
+            EmptyLabelWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+
+
+def test_create_event_group_requires_workflow_context():
+    with pytest.raises(TemporalError):
+        workflow.create_event_group("outside-workflow")
+
+
+####################################################################################################
+# History helpers
+#
+# Markers are rendered as strings so failures print readably and collections can be compared as
+# unordered sets (by sorting). ``_render_marker`` includes the label; ``_render_marker_id`` does
+# not, for the cases where two groups share an id but not a label and the emitted label is
+# unspecified.
+####################################################################################################
+
+
+def _default_marker_id(original_execution_run_id: str, label: str) -> str:
+    return hashlib.sha1(f"{original_execution_run_id}{label}".encode()).hexdigest()
+
+
+def _run_id(handle: WorkflowHandle) -> str:
+    assert handle.first_execution_run_id is not None
+    return handle.first_execution_run_id
+
+
+def _events_of_type(
+    events: Sequence[HistoryEvent], event_type: EventType.ValueType
+) -> list[HistoryEvent]:
+    return [e for e in events if e.event_type == event_type]
+
+
+def _single_event(
+    events: Sequence[HistoryEvent], event_type: EventType.ValueType
+) -> HistoryEvent:
+    matches = _events_of_type(events, event_type)
+    assert (
+        len(matches) == 1
+    ), f"expected 1 {EventType.Name(event_type)}, got {len(matches)}"
+    return matches[0]
+
+
+def _activity_event(events: Sequence[HistoryEvent], activity_id: str) -> HistoryEvent:
+    matches = [
+        e
+        for e in events
+        if e.event_type == EventType.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED
+        and e.activity_task_scheduled_event_attributes.activity_id == activity_id
+    ]
+    assert len(matches) == 1, f"expected 1 activity {activity_id!r}, got {len(matches)}"
+    return matches[0]
+
+
+def _markers_named(
+    events: Sequence[HistoryEvent], marker_name: str
+) -> list[HistoryEvent]:
+    return [
+        e
+        for e in events
+        if e.event_type == EventType.EVENT_TYPE_MARKER_RECORDED
+        and e.marker_recorded_event_attributes.marker_name == marker_name
+    ]
+
+
+def _signaled_event_ids(events: Sequence[HistoryEvent]) -> list[int]:
+    return [
+        e.event_id
+        for e in events
+        if e.event_type == EventType.EVENT_TYPE_WORKFLOW_EXECUTION_SIGNALED
+    ]
+
+
+def _render_marker(marker: EventGroupMarker) -> str:
+    if marker.HasField("inbound_event"):
+        return f"event:{marker.inbound_event.inbound_event_id}"
+    if marker.HasField("inbound_update"):
+        return f"update:{marker.inbound_update.inbound_update_id}"
+    try:
+        label = PayloadConverter.default.from_payload(marker.label.label)
+        return f"label:{marker.label.id}:{label}"
+    except Exception:
+        return f"label:{marker.label.id}"
+
+
+def _render_marker_id(marker: EventGroupMarker) -> str:
+    if marker.HasField("inbound_event"):
+        return f"event:{marker.inbound_event.inbound_event_id}"
+    if marker.HasField("inbound_update"):
+        return f"update:{marker.inbound_update.inbound_update_id}"
+    return f"label:{marker.label.id}"
+
+
+def _markers(event: HistoryEvent) -> list[str]:
+    return sorted(_render_marker(m) for m in event.event_group_markers)
+
+
+def _marker_ids(event: HistoryEvent) -> list[str]:
+    return sorted(_render_marker_id(m) for m in event.event_group_markers)
+
+
+def _assert_markers(event: HistoryEvent, *expected: str) -> None:
+    actual = [_render_marker(m) for m in event.event_group_markers]
+    assert len(actual) == len(
+        expected
+    ), f"marker count {len(actual)} != {len(expected)}: {actual}"
+    assert sorted(actual) == sorted(expected)
+
+
+def _assert_marker_ids(event: HistoryEvent, *expected: str) -> None:
+    actual = [_render_marker_id(m) for m in event.event_group_markers]
+    assert len(actual) == len(
+        expected
+    ), f"marker count {len(actual)} != {len(expected)}: {actual}"
+    assert sorted(actual) == sorted(expected)
+
+
+def _label_marker(group_id: str, label: str) -> str:
+    return f"label:{group_id}:{label}"
+
+
+def _label_marker_id(group_id: str) -> str:
+    return f"label:{group_id}"
+
+
+def _event_marker(event_id: int) -> str:
+    return f"event:{event_id}"
+
+
+def _update_marker(update_id: str) -> str:
+    return f"update:{update_id}"
+
+
+def _label_payload_of(event: HistoryEvent, marker_id: str) -> tuple[str, str]:
+    for marker in event.event_group_markers:
+        if marker.HasField("label") and marker.label.id == marker_id:
+            encoding = marker.label.label.metadata["encoding"].decode()
+            return encoding, marker.label.label.data.decode()
+    raise AssertionError(f"no label marker {marker_id!r} on event")
+
+
+def _raw_label_payload(event: HistoryEvent, marker_id: str) -> Payload:
+    for marker in event.event_group_markers:
+        if marker.HasField("label") and marker.label.id == marker_id:
+            return marker.label.label
+    raise AssertionError(f"no label marker {marker_id!r} on event")
+
+
+async def _fetch_events(handle: WorkflowHandle) -> list[HistoryEvent]:
+    return list((await handle.fetch_history()).events)
+
+
+# Module-level so every workflow can issue a uniquely keyed activity without repeating options.
+async def _activity(
+    activity_id: str,
+    event_groups: Sequence[workflow.EventGroup] | None = None,
+) -> None:
+    await workflow.execute_activity(
+        noop_activity,
+        start_to_close_timeout=_ACT_TIMEOUT,
+        activity_id=activity_id,
+        event_groups=event_groups,
+    )
+
+
+async def _swallow(aw: object) -> None:
+    try:
+        await aw  # type: ignore[misc]
+    except (
+        asyncio.CancelledError,
+        asyncio.TimeoutError,
+        ActivityError,
+        ChildWorkflowError,
+        NexusOperationError,
+    ):
+        pass
+
+
+@activity.defn
+async def noop_activity() -> None:
+    return None
+
+
+@activity.defn
+async def control_activity(value: str) -> str:
+    return value
+
+
+@activity.defn
+async def sleep_activity() -> None:
+    await asyncio.sleep(5)
+
+
+@activity.defn
+async def fail_first_activity() -> None:
+    if activity.info().attempt == 1:
+        raise ApplicationError("retry me")
+
+
+@workflow.defn
+class NoopChildWorkflow:
+    @workflow.run
+    async def run(self) -> None:
+        return None
+
+
+@workflow.defn
+class SleepChildWorkflow:
+    @workflow.run
+    async def run(self) -> None:
+        await workflow.sleep(5)