Add erlang.spawn_task() for fire-and-forget async tasks (#22)

* Add erlang.spawn_task() for fire-and-forget async tasks

Implements spawn_task() function that creates and schedules async tasks
from both sync and async contexts. This is useful for spawning background
work from synchronous Python code called by Erlang.

The function:
- Returns an asyncio.Task for optional await/cancel
- Automatically wakes up the event loop in sync context
- Works with both ErlangEventLoop and standard asyncio loops

* Add spawn_task to changelog and asyncio documentation

- Add Unreleased section with spawn_task feature
- Document erlang.spawn_task() in asyncio.md API reference

Author: benoitc

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog
 
+## Unreleased
+
+### Added
+
+- **`erlang.spawn_task(coro)`** - Spawn async tasks from both sync and async contexts
+  - Works in sync code called by Erlang (where `asyncio.get_running_loop()` fails)
+  - Returns `asyncio.Task` for optional await/cancel (fire-and-forget pattern)
+  - Automatically wakes up the event loop in sync context
+  - Works with both ErlangEventLoop and standard asyncio loops
+
 ## 2.0.0 (2026-03-09)
 
 ### Added

docs/asyncio.md

@@ -784,6 +784,40 @@ async def main():
 erlang.run(main())
 ```
 
+#### erlang.spawn_task(coro, *, name=None)
+
+Spawn an async task from both sync and async contexts. This is useful for fire-and-forget background work from synchronous Python code called by Erlang.
+
+```python
+import erlang
+
+# From sync code called by Erlang
+def handle_request(data):
+    # This works even though there's no running event loop
+    erlang.spawn_task(process_async(data))
+    return 'ok'
+
+# From async code
+async def handler():
+    # Also works in async context
+    erlang.spawn_task(background_work())
+    await other_work()
+
+async def process_async(data):
+    await asyncio.sleep(0.1)
+    # Do async processing...
+
+async def background_work():
+    await asyncio.sleep(0.1)
+    # Do background work...
+```
+
+**Key features:**
+- Works in sync context where `asyncio.get_running_loop()` would fail
+- Returns `asyncio.Task` for optional await/cancel
+- Automatically wakes up the event loop to ensure the task runs promptly
+- Works with both ErlangEventLoop and standard asyncio loops
+
 #### asyncio.wait(fs, *, timeout=None, return_when=ALL_COMPLETED)
 
 Wait for multiple futures/tasks.

priv/_erlang_impl/__init__.py

@@ -66,6 +66,7 @@
 
 __all__ = [
     'run',
+    'spawn_task',
     'new_event_loop',
     'get_event_loop_policy',
     'install',
@@ -162,6 +163,89 @@ async def main():
                 loop.close()
 
 
+def spawn_task(coro, *, name=None):
+    """Spawn an async task, working in both async and sync contexts.
+
+    This function creates and schedules a task on the event loop, with
+    automatic wakeup for Erlang-driven loops where the loop may not be
+    actively polling.
+
+    Args:
+        coro: The coroutine to run as a task.
+        name: Optional name for the task (Python 3.8+).
+
+    Returns:
+        asyncio.Task: The created task. Can be ignored (fire-and-forget)
+        or awaited/cancelled if needed.
+
+    Raises:
+        RuntimeError: If no event loop is available or the loop is closed.
+
+    Example:
+        # From sync code called by Erlang
+        def handle_request(data):
+            erlang.spawn_task(process_async(data))
+            return 'ok'
+
+        # From async code
+        async def handler():
+            erlang.spawn_task(background_work())
+            await other_work()
+    """
+    # Try to get the running loop first (works in async context)
+    try:
+        loop = asyncio.get_running_loop()
+        # In async context, just create_task directly
+        if name is not None:
+            return loop.create_task(coro, name=name)
+        else:
+            return loop.create_task(coro)
+    except RuntimeError:
+        pass
+
+    # Sync context: get the event loop
+    try:
+        loop = asyncio.get_event_loop()
+    except RuntimeError:
+        coro.close()  # Prevent "coroutine was never awaited" warning
+        raise RuntimeError(
+            "No event loop available. Ensure erlang is initialized or "
+            "call from within an async context."
+        )
+
+    if loop.is_closed():
+        coro.close()
+        raise RuntimeError("Event loop is closed")
+
+    # Create the task
+    try:
+        if name is not None:
+            task = loop.create_task(coro, name=name)
+        else:
+            task = loop.create_task(coro)
+    except Exception:
+        coro.close()
+        raise
+
+    # Wake up the event loop to process the task
+    # This is critical for sync context - without wakeup, the task
+    # waits until the next event/timeout
+    if hasattr(loop, '_pel') and hasattr(loop, '_loop_capsule'):
+        # ErlangEventLoop - use native wakeup
+        try:
+            loop._pel._wakeup_for(loop._loop_capsule)
+        except Exception:
+            pass
+    elif hasattr(loop, '_write_to_self'):
+        # Standard asyncio loop - use self-pipe trick
+        try:
+            loop._write_to_self()
+        except Exception:
+            pass
+
+    return task
+
+
 def install():
     """Install ErlangEventLoopPolicy as the default event loop policy.
 

priv/tests/test_spawn_task.py

@@ -0,0 +1,113 @@
+# Copyright 2026 Benoit Chesneau
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+Tests for erlang.spawn_task() function.
+
+Tests spawning async tasks from both sync and async contexts.
+"""
+
+import asyncio
+import os
+import sys
+import unittest
+
+from . import _testbase as tb
+
+
+def _get_spawn_task():
+    """Get spawn_task function, handling import path setup."""
+    try:
+        import erlang
+        return erlang.spawn_task
+    except ImportError:
+        pass
+
+    # Add priv directory to path for _erlang_impl
+    priv_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+    if priv_dir not in sys.path:
+        sys.path.insert(0, priv_dir)
+
+    from _erlang_impl import spawn_task
+    return spawn_task
+
+
+spawn_task = _get_spawn_task()
+
+
+class _TestSpawnTask:
+    """Tests for spawn_task functionality."""
+
+    def test_spawn_task_from_async_context(self):
+        """Test spawning a task from async context."""
+        results = []
+
+        async def background():
+            results.append('background')
+
+        async def main():
+            task = spawn_task(background())
+            self.assertIsInstance(task, asyncio.Task)
+            await asyncio.sleep(0.01)
+            results.append('main')
+
+        self.loop.run_until_complete(main())
+        self.assertIn('background', results)
+        self.assertIn('main', results)
+
+    def test_spawn_task_returns_awaitable(self):
+        """Test that spawn_task returns an awaitable Task."""
+        async def compute():
+            return 42
+
+        async def main():
+            task = spawn_task(compute())
+            result = await task
+            self.assertEqual(result, 42)
+
+        self.loop.run_until_complete(main())
+
+    def test_spawn_task_with_name(self):
+        """Test spawning a named task."""
+        async def noop():
+            pass
+
+        async def main():
+            task = spawn_task(noop(), name='test-task')
+            self.assertEqual(task.get_name(), 'test-task')
+            await task
+
+        self.loop.run_until_complete(main())
+
+    def test_spawn_task_exception_handling(self):
+        """Test that exceptions in spawned tasks are captured."""
+        async def failing():
+            raise ValueError("test error")
+
+        async def main():
+            task = spawn_task(failing())
+            await asyncio.sleep(0.01)
+            self.assertTrue(task.done())
+            with self.assertRaises(ValueError):
+                task.result()
+
+        self.loop.run_until_complete(main())
+
+
+class TestErlangSpawnTask(_TestSpawnTask, tb.ErlangTestCase):
+    pass
+
+
+class TestAIOSpawnTask(_TestSpawnTask, tb.AIOTestCase):
+    pass