@@ -0,0 +1 @@
+Core: Worker heartbeats now report correct task-slot and poller counts when using buffered or custom metrics backends, including the Python SDK's `MetricBuffer`. Previously these counts could remain `0` because heartbeat accounting depended on reading labels from backend-owned attributes.
@@ -169,6 +169,10 @@ async def env(env_type: str) -> AsyncGenerator[WorkflowEnvironment, None]:
                 "--dynamic-config-value",
                 "frontend.enableCancelWorkerPollsOnShutdown=true",
                 "--dynamic-config-value",
+                "frontend.WorkerHeartbeatsEnabled=true",
+                "--dynamic-config-value",
+                "frontend.ListWorkersEnabled=true",
+                "--dynamic-config-value",
                 "component.nexusoperations.recordCancelRequestCompletionEvents=true",
                 "--dynamic-config-value",
                 "activity.enableStandalone=true",
 
@@ -36,6 +36,7 @@
 
 import temporalio.activity
 import temporalio.api.sdk.v1
+import temporalio.api.workflowservice.v1
 import temporalio.client
 import temporalio.converter
 import temporalio.converter._extstore
@@ -5489,6 +5490,54 @@ async def test_workflow_buffered_metrics(client: Client, env: WorkflowEnvironmen
     )
 
 
+@pytest.mark.requires_local_server
+async def test_workflow_buffered_metrics_worker_heartbeat(env: WorkflowEnvironment):
+    if env.supports_time_skipping:
+        pytest.skip("Time-skipping server does not implement ListWorkers")
+
+    runtime = Runtime(
+        telemetry=TelemetryConfig(metrics=MetricBuffer(10000)),
+        worker_heartbeat_interval=timedelta(seconds=1),
+    )
+    client = await env.connect_client(runtime=runtime)
+    async with new_worker(
+        client, CustomMetricsWorkflow, activities=[custom_metrics_activity]
+    ) as worker:
+        await client.execute_workflow(
+            CustomMetricsWorkflow.run,
+            id=f"wf-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+
+        async def assert_worker_heartbeat() -> None:
+            response = await client.workflow_service.list_workers(
+                temporalio.api.workflowservice.v1.ListWorkersRequest(
+                    namespace=client.namespace,
+                    query=f'TaskQueue = "{worker.task_queue}"',
+                    page_size=100,
+                )
+            )
+            heartbeat = next(
+                (
+                    info.worker_heartbeat
+                    for info in response.workers_info
+                    if info.worker_heartbeat.task_queue == worker.task_queue
+                ),
+                None,
+            )
+            assert heartbeat is not None
+            assert heartbeat.workflow_task_slots_info.total_processed_tasks > 0
+            assert heartbeat.activity_task_slots_info.total_processed_tasks > 0
+            assert heartbeat.workflow_task_slots_info.current_available_slots > 0
+            assert heartbeat.activity_task_slots_info.current_available_slots > 0
+            assert heartbeat.local_activity_slots_info.current_available_slots > 0
+            assert heartbeat.workflow_poller_info.current_pollers > 0
+            assert heartbeat.activity_poller_info.current_pollers > 0
+
+        # Do not drain the buffer: heartbeats must not depend on initializing Python attributes.
+        await assert_eventually(assert_worker_heartbeat)
+
+
 async def test_workflow_metrics_other_types(env: WorkflowEnvironment):
     async def do_stuff(buffer: MetricBuffer) -> float:
         runtime = Runtime(telemetry=TelemetryConfig(metrics=buffer))
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1 @@`
	`1`	+Core: Worker heartbeats now report correct task-slot and poller counts when using buffered or custom metrics backends, including the Python SDK's `MetricBuffer`. Previously these counts could remain `0` because heartbeat accounting depended on reading labels from backend-owned attributes.