Fix service nursery shielding when TaskStatus.started() is called from outside the new task

newsfragments/27.bugfix.rst

@@ -0,0 +1,6 @@
+:func:`open_service_nursery` no longer assumes that ``TaskStatus.started()``
+will be called from inside the task that was just started. This restores
+feature parity with regular Trio nurseries, which allow ``started()`` to be
+called anywhere, and fixes
+`trio-asyncio issue #135 <https://github.com/python-trio/trio-asyncio/issues/135>`__.
+

tricycle/_service_nursery.py

@@ -114,10 +114,11 @@ async def wrap_child(*, task_status: trio.TaskStatus[Any]) -> None:
                 # For start(), the child doesn't get shielded until it
                 # calls task_status.started().
                 shield_scope = child_task_scopes.open_child(shield=False)
+                child_task = trio.lowlevel.current_task()
 
                 def wrap_started(value: object = None) -> None:
                     type(task_status).started(task_status, value)  # type: ignore
-                    if trio.lowlevel.current_task().parent_nursery is not nursery:
+                    if child_task.parent_nursery is not nursery:
                         # started() didn't move the task due to a cancellation,
                         # so it doesn't get the shield
                         return

tricycle/_tests/test_service_nursery.py

@@ -79,6 +79,37 @@ async def shielded_sleep_then_start(*, task_status: trio.TaskStatus[None]) -> No
     assert record == ["parent task finished", "background task exiting"]
 
 
+async def test_remote_start(autojump_clock: trio.testing.MockClock) -> None:
+    record = []
+    outer_task_status: trio.TaskStatus[int]
+
+    async def task(*, task_status: trio.TaskStatus[int]) -> None:
+        nonlocal outer_task_status
+        outer_task_status = task_status
+        try:
+            await trio.sleep(10)
+            record.append("background task finished")  # pragma: no cover
+        finally:
+            record.append("background task exiting")
+
+    async def delayed_start() -> None:
+        await trio.sleep(1)
+        outer_task_status.started(42)
+
+    async with trio.open_nursery() as outer_nursery:
+        outer_nursery.start_soon(delayed_start)
+        async with open_service_nursery() as inner_nursery:
+            assert 42 == await inner_nursery.start(task)
+            assert trio.current_time() == 1.0
+            outer_nursery.cancel_scope.cancel()
+            with trio.CancelScope(shield=True):
+                await trio.sleep(1)
+            record.append("parent task finished")
+
+    assert trio.current_time() == 2.0
+    assert record == ["parent task finished", "background task exiting"]
+
+
 async def test_problems() -> None:
     async with open_service_nursery() as nursery:
         with pytest.raises(TypeError) as info: