Handle case where in_workflow is called in a synchronous activity (#884)

* Handle case where in_workflow is called in a synchronous activity

* Formatting

* Move the catch down into

* Add value validation

* Format after merge

Author: tconley1428

temporalio/workflow.py

@@ -642,7 +642,12 @@ def current() -> _Runtime:
 
     @staticmethod
     def maybe_current() -> Optional[_Runtime]:
-        return getattr(asyncio.get_running_loop(), "__temporal_workflow_runtime", None)
+        try:
+            return getattr(
+                asyncio.get_running_loop(), "__temporal_workflow_runtime", None
+            )
+        except RuntimeError:
+            return None
 
     @staticmethod
     def set_on_loop(

tests/worker/test_workflow.py

@@ -7975,6 +7975,37 @@ async def test_quick_activity_swallows_cancellation(client: Client):
         temporalio.worker._workflow_instance._raise_on_cancelling_completed_activity_override = False
 
 
+@activity.defn
+def use_in_workflow() -> bool:
+    return workflow.in_workflow()
+
+
+@workflow.defn
+class UseInWorkflow:
+    @workflow.run
+    async def run(self):
+        res = await workflow.execute_activity(
+            use_in_workflow, schedule_to_close_timeout=timedelta(seconds=10)
+        )
+        return res
+
+
+async def test_in_workflow_sync(client: Client):
+    async with new_worker(
+        client,
+        UseInWorkflow,
+        activities=[use_in_workflow],
+        activity_executor=concurrent.futures.ThreadPoolExecutor(max_workers=1),
+    ) as worker:
+        res = await client.execute_workflow(
+            UseInWorkflow.run,
+            id=f"test_in_workflow_sync",
+            task_queue=worker.task_queue,
+            execution_timeout=timedelta(minutes=1),
+        )
+        assert not res
+
+
 class SignalInterceptor(temporalio.worker.Interceptor):
     def workflow_interceptor_class(
         self, input: temporalio.worker.WorkflowInterceptorClassInput