Add TaskGroup.start() and TaskStatus for task readiness signaling

Add TaskStatus class with started() method and TaskGroup.start()
async method. The task calls task_status.started(value) to signal
it is ready, and start() returns that value to the caller. The task
continues running in the group after signaling readiness.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: kovan

Lib/asyncio/taskgroups.py

@@ -2,14 +2,40 @@
 # license: PSFL.
 
 
-__all__ = ("TaskGroup",)
+__all__ = ("TaskGroup", "TaskStatus")
 
 from . import events
 from . import exceptions
 from . import futures
 from . import tasks
 
 
+class TaskStatus:
+    """Status object passed to tasks started via :meth:`TaskGroup.start`.
+
+    The task calls :meth:`started` to signal readiness, passing an
+    optional value back to the ``start()`` caller.
+    """
+
+    def __init__(self):
+        self._future = None   # set by TaskGroup.start()
+        self._started = False
+
+    def started(self, value=None):
+        """Signal that the task is ready.
+
+        *value* is returned to the ``await TaskGroup.start(...)`` caller.
+        May only be called once.
+        """
+        if self._started:
+            raise RuntimeError("task already signalled readiness")
+        if self._future is None or self._future.done():
+            raise RuntimeError(
+                "TaskStatus is not associated with a pending start()")
+        self._started = True
+        self._future.set_result(value)
+
+
 class TaskGroup:
     """Asynchronous context manager for managing groups of tasks.
 
@@ -210,6 +236,39 @@ def create_task(self, coro, **kwargs):
             # task.exception().__traceback__->TaskGroup.create_task->task
             del task
 
+    async def start(self, coro_fn, *args, name=None, context=None):
+        """Start a task and wait until it signals readiness.
+
+        *coro_fn* is called as ``coro_fn(*args, task_status=task_status)``.
+        The coroutine must call ``task_status.started(value)`` to signal
+        readiness.  The *value* passed to ``started()`` is returned by
+        this method.  The task continues running in the group after
+        ``started()`` is called.
+        """
+        if not self._entered:
+            raise RuntimeError(f"TaskGroup {self!r} has not been entered")
+        if self._exiting and not self._tasks:
+            raise RuntimeError(f"TaskGroup {self!r} is finished")
+        if self._aborting:
+            raise RuntimeError(f"TaskGroup {self!r} is shutting down")
+
+        task_status = TaskStatus()
+        task_status._future = self._loop.create_future()
+
+        coro = coro_fn(*args, task_status=task_status)
+        kwargs = {}
+        if name is not None:
+            kwargs['name'] = name
+        if context is not None:
+            kwargs['context'] = context
+        task = self.create_task(coro, **kwargs)
+
+        try:
+            return await task_status._future
+        except BaseException:
+            task.cancel()
+            raise
+
     # Since Python 3.8 Tasks propagate all exceptions correctly,
     # except for KeyboardInterrupt and SystemExit which are
     # still considered special.

Lib/test/test_asyncio/test_taskgroups.py

@@ -1103,6 +1103,108 @@ async def throw_error():
         await asyncio.sleep(0)
 
 
+    # -- TaskGroup.start() / TaskStatus ------------------------------------
+
+    async def test_start_basic(self):
+        """start() returns the value passed to task_status.started()."""
+        async def server(task_status):
+            task_status.started(42)
+            await asyncio.sleep(0)
+
+        async with taskgroups.TaskGroup() as tg:
+            value = await tg.start(server)
+        self.assertEqual(value, 42)
+
+    async def test_start_none_value(self):
+        """started() with no arg returns None."""
+        async def worker(task_status):
+            task_status.started()
+            await asyncio.sleep(0)
+
+        async with taskgroups.TaskGroup() as tg:
+            value = await tg.start(worker)
+        self.assertIsNone(value)
+
+    async def test_start_task_continues(self):
+        """Task keeps running after started() is called."""
+        finished = False
+
+        async def worker(task_status):
+            nonlocal finished
+            task_status.started("ready")
+            await asyncio.sleep(0)
+            finished = True
+
+        async with taskgroups.TaskGroup() as tg:
+            value = await tg.start(worker)
+            self.assertEqual(value, "ready")
+        # TaskGroup waits for the task to finish
+        self.assertTrue(finished)
+
+    async def test_start_error_before_started(self):
+        """Exception before started() propagates through the group."""
+        async def failing(task_status):
+            raise RuntimeError("boom")
+
+        with self.assertRaises(ExceptionGroup) as cm:
+            async with taskgroups.TaskGroup() as tg:
+                await tg.start(failing)
+
+        self.assertEqual(len(cm.exception.exceptions), 1)
+        self.assertIsInstance(cm.exception.exceptions[0], RuntimeError)
+
+    async def test_start_cancelled_before_started(self):
+        """If the task is cancelled before started(), cancellation propagates."""
+        async def slow_start(task_status):
+            await asyncio.sleep(100)
+            task_status.started()
+
+        with self.assertRaises(TimeoutError):
+            async with asyncio.timeout(0.01):
+                async with taskgroups.TaskGroup() as tg:
+                    await tg.start(slow_start)
+
+    async def test_start_already_started_error(self):
+        """Calling started() twice raises RuntimeError."""
+        async def double_start(task_status):
+            task_status.started(1)
+            with self.assertRaises(RuntimeError):
+                task_status.started(2)
+
+        async with taskgroups.TaskGroup() as tg:
+            value = await tg.start(double_start)
+        self.assertEqual(value, 1)
+
+    async def test_start_multiple_tasks(self):
+        """Multiple start() calls in the same group."""
+        async def worker(n, task_status):
+            task_status.started(n * 10)
+            await asyncio.sleep(0)
+
+        async with taskgroups.TaskGroup() as tg:
+            v1 = await tg.start(worker, 1)
+            v2 = await tg.start(worker, 2)
+            v3 = await tg.start(worker, 3)
+
+        self.assertEqual(v1, 10)
+        self.assertEqual(v2, 20)
+        self.assertEqual(v3, 30)
+
+    async def test_start_with_name(self):
+        """start() passes name= to create_task."""
+        task_name = None
+
+        async def worker(task_status):
+            nonlocal task_name
+            task_name = asyncio.current_task().get_name()
+            task_status.started()
+
+        async with taskgroups.TaskGroup() as tg:
+            await tg.start(worker, name="my-worker")
+
+        self.assertEqual(task_name, "my-worker")
+
+
 class TestTaskGroup(BaseTestTaskGroup, unittest.IsolatedAsyncioTestCase):
     loop_factory = asyncio.EventLoop