Use the activity info provided during interceptor construction (#903)

* Use the activity info provided during interceptor construction instead of the activity context

* Linting

Author: tconley1428

temporalio/worker/_activity.py

@@ -784,8 +784,7 @@ def info(self) -> temporalio.activity.Info:
         return self._info
 
     def heartbeat(self, *details: Any) -> None:
-        info = temporalio.activity.info()
-        self._worker._heartbeat(info.task_token, *details)
+        self._worker._heartbeat(self._info.task_token, *details)
 
 
 # This has to be defined at the top-level to be picklable for process executors

tests/worker/test_activity.py

@@ -1416,3 +1416,30 @@ async def raise_error():
 
     finally:
         activity.logger.base_logger.removeHandler(CustomLogHandler())
+
+
+async def test_activity_heartbeat_context(client: Client, worker: ExternalWorker):
+    @activity.defn
+    async def heartbeat():
+        if activity.info().attempt == 1:
+            context: activity._Context = activity._Context.current()
+
+            def heartbeat_task():
+                async def h():
+                    if context.heartbeat is not None:
+                        context.heartbeat("Some detail")
+
+                asyncio.run(h())
+
+            thread = threading.Thread(target=heartbeat_task)
+            thread.start()
+            thread.join()
+            raise RuntimeError("oh no!")
+        else:
+            assert len(activity.info().heartbeat_details) == 1
+            return "details: " + activity.info().heartbeat_details[0]
+
+    result = await _execute_workflow_with_activity(
+        client, worker, heartbeat, retry_max_attempts=2
+    )
+    assert result.result == "details: Some detail"