Repository: agronholm/anyio

Issue:
parentage of `await tg.start(` tasks incorrect on asyncio backend
demo program:

```python
import anyio
import trio.lowlevel

def print_parent_task():
    try:
        print(f"{trio.lowlevel.current_task().parent_nursery.parent_task=}")
    except RuntimeError:
        pass

async def server(*, task_status):
    print(f"{anyio.get_current_task().parent_id=}")
    print_parent_task()
    await anyio.sleep(0.1)
    task_status.started()
    print(f"{anyio.get_current_task().parent_id=}")
    print_parent_task()
    await anyio.sleep(0.1)
    print(f"{anyio.get_current_task().parent_id=}")
    print_parent_task()


async def other(tg):
    async with anyio.create_task_group() as other_tg:
        other_tg.start_soon(tg.start, server)


async def amain():
    async with anyio.create_task_group() as tg:
        tg.start_soon(other, tg)


print("=========== trio ============")
anyio.run(amain, backend="trio")
print("=========== asyncio ============")
anyio.run(amain, backend="asyncio")
```

output:

```python
=========== trio ============
anyio.get_current_task().parent_id=140390771298112
trio.lowlevel.current_task().parent_nursery.parent_task=<Task 'anyio._backends._trio.TaskGroup.start' at 0x7faf460d8f40>
anyio.get_current_task().parent_id=140390771297632
trio.lowlevel.current_task().parent_nursery.parent_task=<Task '__main__.amain' at 0x7faf460d8d60>
anyio.get_current_task().parent_id=140390771297632
trio.lowlevel.current_task().parent_nursery.parent_task=<Task '__main__.amain' at 0x7faf460d8d60>
=========== asyncio ============
anyio.get_current_task().parent_id=140390780426336
anyio.get_current_task().parent_id=140390780426336
anyio.get_current_task().parent_id=140390780426336
```

Discussion:

Patch:
diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst
index d43a3f1..c07f26f 100644
--- a/docs/versionhistory.rst
+++ b/docs/versionhistory.rst
@@ -12,6 +12,7 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
   ``BlockingPortal.wrap_async_context_manager()`` if the host task is cancelled
   (`#381 <https://github.com/agronholm/anyio/issues/381>`_; PR by Jonathan Slenders)
 - Fixed worker threads being marked as being event loop threads in sniffio
+- Fixed task parent ID not getting set to the correct value on asyncio
 
 **3.3.4**
 
diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py
index 834dd91..3e9e8d2 100644
--- a/src/anyio/_backends/_asyncio.py
+++ b/src/anyio/_backends/_asyncio.py
@@ -515,8 +515,9 @@ class ExceptionGroup(BaseExceptionGroup):
 
 
 class _AsyncioTaskStatus(abc.TaskStatus):
-    def __init__(self, future: asyncio.Future):
+    def __init__(self, future: asyncio.Future, parent_id: int):
         self._future = future
+        self._parent_id = parent_id
 
     def started(self, value: object = None) -> None:
         try:
@@ -524,6 +525,9 @@ class _AsyncioTaskStatus(abc.TaskStatus):
         except asyncio.InvalidStateError:
             raise RuntimeError("called 'started' twice on the same task status") from None
 
+        task = cast(asyncio.Task, current_task())
+        _task_states[task].parent_id = self._parent_id
+
 
 class TaskGroup(abc.TaskGroup):
     def __init__(self) -> None:
@@ -653,7 +657,11 @@ class TaskGroup(abc.TaskGroup):
 
         kwargs = {}
         if task_status_future:
-            kwargs['task_status'] = _AsyncioTaskStatus(task_status_future)
+            parent_id = id(current_task())
+            kwargs['task_status'] = _AsyncioTaskStatus(task_status_future,
+                                                       id(self.cancel_scope._host_task))
+        else:
+            parent_id = id(self.cancel_scope._host_task)
 
         coro = func(*args, **kwargs)
         if not asyncio.iscoroutine(coro):
@@ -668,7 +676,7 @@ class TaskGroup(abc.TaskGroup):
             task.add_done_callback(task_done)
 
         # Make the spawned task inherit the task group's cancel scope
-        _task_states[task] = TaskState(parent_id=id(current_task()), name=name,
+        _task_states[task] = TaskState(parent_id=parent_id, name=name,
                                        cancel_scope=self.cancel_scope)
         self.cancel_scope._tasks.add(task)
         return task


Test patch:
diff --git a/tests/test_taskgroups.py b/tests/test_taskgroups.py
index 1980470..fd60344 100644
--- a/tests/test_taskgroups.py
+++ b/tests/test_taskgroups.py
@@ -2,7 +2,7 @@ import asyncio
 import re
 import sys
 import time
-from typing import Any, AsyncGenerator, Coroutine, Dict, Generator, NoReturn, Set
+from typing import Any, AsyncGenerator, Coroutine, Dict, Generator, NoReturn, Optional, Set
 
 import pytest
 import trio
@@ -10,7 +10,8 @@ import trio
 import anyio
 from anyio import (
     CancelScope, ExceptionGroup, create_task_group, current_effective_deadline, current_time,
-    fail_after, get_cancelled_exc_class, move_on_after, sleep, wait_all_tasks_blocked)
+    fail_after, get_cancelled_exc_class, get_current_task, move_on_after, sleep,
+    wait_all_tasks_blocked)
 from anyio.abc import TaskGroup, TaskStatus
 from anyio.lowlevel import checkpoint
 
@@ -1004,3 +1005,45 @@ async def test_cancellederror_combination_with_message() -> None:
             await wait_all_tasks_blocked()
             assert isinstance(task, asyncio.Task)
             task.cancel('blah')
+
+
+async def test_start_soon_parent_id() -> None:
+    root_task_id = get_current_task().id
+    parent_id: Optional[int] = None
+
+    async def subtask() -> None:
+        nonlocal parent_id
+        parent_id = get_current_task().parent_id
+
+    async def starter_task() -> None:
+        tg.start_soon(subtask)
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(starter_task)
+
+    assert parent_id == root_task_id
+
+
+async def test_start_parent_id() -> None:
+    root_task_id = get_current_task().id
+    starter_task_id: Optional[int] = None
+    initial_parent_id: Optional[int] = None
+    permanent_parent_id: Optional[int] = None
+
+    async def subtask(*, task_status: TaskStatus) -> None:
+        nonlocal initial_parent_id, permanent_parent_id
+        initial_parent_id = get_current_task().parent_id
+        task_status.started()
+        permanent_parent_id = get_current_task().parent_id
+
+    async def starter_task() -> None:
+        nonlocal starter_task_id
+        starter_task_id = get_current_task().id
+        await tg.start(subtask)
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(starter_task)
+
+    assert initial_parent_id != permanent_parent_id
+    assert initial_parent_id == starter_task_id
+    assert permanent_parent_id == root_task_id