add a integration test

Signed-off-by: Tim Li <ltim@uber.com>

Author: timl3136

tests/cadence/_internal/activity/test_activity_executor.py

@@ -451,3 +451,61 @@ async def activity_fn():
             identity="identity",
         )
     )
+
+
+async def test_heartbeat_details_recovery_across_attempts(client):
+    """Simulate retry: first attempt has no heartbeat details and fails,
+    second attempt receives heartbeat details from the server and succeeds."""
+    worker_stub = client.worker_stub
+    worker_stub.RespondActivityTaskFailed = AsyncMock(
+        return_value=RespondActivityTaskFailedResponse()
+    )
+    worker_stub.RespondActivityTaskCompleted = AsyncMock(
+        return_value=RespondActivityTaskCompletedResponse()
+    )
+    worker_stub.RecordActivityTaskHeartbeat = AsyncMock(
+        return_value=RecordActivityTaskHeartbeatResponse()
+    )
+
+    attempt_count = 0
+
+    reg = Registry()
+
+    @reg.activity(name="activity_type")
+    async def activity_fn():
+        nonlocal attempt_count
+        attempt_count += 1
+
+        details = activity.heartbeat_details()
+        if not details:
+            activity.heartbeat("step1", 50)
+            raise RuntimeError("simulated failure on first attempt")
+
+        return activity.heartbeat_details(str, int)
+
+    executor = ActivityExecutor(client, "task_list", "identity", 1, reg.get_activity)
+
+    # First attempt: no heartbeat details, activity heartbeats progress then fails
+    await executor.execute(fake_task("activity_type", ""))
+    worker_stub.RespondActivityTaskFailed.assert_called_once()
+    worker_stub.RecordActivityTaskHeartbeat.assert_called_once_with(
+        RecordActivityTaskHeartbeatRequest(
+            task_token=b"task_token",
+            details=Payload(data=b'"step1" 50'),
+            identity="identity",
+        )
+    )
+
+    # Second attempt: server provides heartbeat details from previous attempt
+    await executor.execute(
+        fake_task("activity_type", "", heartbeat_details='"step1" 50')
+    )
+    worker_stub.RespondActivityTaskCompleted.assert_called_once_with(
+        RespondActivityTaskCompletedRequest(
+            task_token=b"task_token",
+            result=Payload(data=b'["step1",50]'),
+            identity="identity",
+        )
+    )
+
+    assert attempt_count == 2