diff --git a/.coveragerc b/.coveragerc
index 431a02971b..5272237caf 100644
--- a/.coveragerc
+++ b/.coveragerc
@@ -23,6 +23,7 @@ exclude_lines =
   if _t.TYPE_CHECKING:
   if t.TYPE_CHECKING:
   @overload
+  class .*\bProtocol\b.*\):
 
 partial_branches =
   pragma: no branch
diff --git a/docs/source/conf.py b/docs/source/conf.py
index 39eda12ad5..06e4a7e6be 100755
--- a/docs/source/conf.py
+++ b/docs/source/conf.py
@@ -62,12 +62,16 @@
     ("py:obj", "trio._abc.SendType"),
     ("py:obj", "trio._abc.T"),
     ("py:obj", "trio._abc.T_resource"),
+    ("py:class", "trio._core._run.StatusT"),
+    ("py:class", "trio._core._run.StatusT_co"),
+    ("py:class", "trio._core._run.StatusT_contra"),
+    ("py:class", "trio._core._run.RetT"),
     ("py:class", "trio._threads.T"),
-    # why aren't these found in stdlib?
-    ("py:class", "types.FrameType"),
     ("py:class", "P.args"),
     ("py:class", "P.kwargs"),
     ("py:class", "RetT"),
+    # why aren't these found in stdlib?
+    ("py:class", "types.FrameType"),
     # TODO: figure out if you can link this to SSL
     ("py:class", "Context"),
     # TODO: temporary type
@@ -90,6 +94,24 @@
 }
 
 
+def autodoc_process_signature(
+    app, what, name, obj, options, signature, return_annotation
+):
+    """Modify found signatures to fix various issues."""
+    if signature is not None:
+        signature = signature.replace("~_contextvars.Context", "~contextvars.Context")
+        if name == "trio.lowlevel.start_guest_run":
+            signature = signature.replace("Outcome", "~outcome.Outcome")
+        if name == "trio.lowlevel.RunVar":  # Typevar is not useful here.
+            signature = signature.replace(": ~trio._core._local.T", "")
+        if "_NoValue" in signature:
+            # Strip the type from the union, make it look like = ...
+            signature = signature.replace(" | type[trio._core._local._NoValue]", "")
+            signature = signature.replace("<class 'trio._core._local._NoValue'>", "...")
+
+    return signature, return_annotation
+
+
 # XX hack the RTD theme until
 #   https://github.com/rtfd/sphinx_rtd_theme/pull/382
 # is shipped (should be in the release after 0.2.4)
@@ -97,6 +119,7 @@
 # though.
 def setup(app):
     app.add_css_file("hackrtd.css")
+    app.connect("autodoc-process-signature", autodoc_process_signature)
 
 
 # -- General configuration ------------------------------------------------
diff --git a/docs/source/reference-core.rst b/docs/source/reference-core.rst
index 30994e4386..316a9c2ac3 100644
--- a/docs/source/reference-core.rst
+++ b/docs/source/reference-core.rst
@@ -916,11 +916,19 @@ The nursery API
 
 
 .. autoclass:: Nursery()
-   :members:
+   :members: child_tasks, parent_task
+
+   .. automethod:: start(async_fn, *args, name = None)
+
+   .. automethod:: start_soon(async_fn, *args, name = None)
 
 .. attribute:: TASK_STATUS_IGNORED
+   :type: TaskStatus
 
-   See :meth:`~Nursery.start`.
+   See :meth:`Nursery.start`.
+
+.. autoclass:: TaskStatus(Protocol[StatusT])
+   :members:
 
 .. _task-local-storage:
 
diff --git a/docs/source/reference-lowlevel.rst b/docs/source/reference-lowlevel.rst
index faf07268cb..712a36ad04 100644
--- a/docs/source/reference-lowlevel.rst
+++ b/docs/source/reference-lowlevel.rst
@@ -49,7 +49,11 @@ attributes, :meth:`trio.Lock.statistics`, etc.). Here are some more.
 Global statistics
 -----------------
 
-.. autofunction:: current_statistics
+.. function:: current_statistics() -> RunStatistics
+
+   Returns an object containing run-loop-level debugging information:
+
+.. autoclass:: RunStatistics()
 
 
 The current clock
@@ -534,9 +538,6 @@ Task API
       putting a task to sleep and then waking it up again. (See
       :func:`wait_task_rescheduled` for details.)
 
-.. autoclass:: TaskStatus
-   :members:
-
 .. _guest-mode:
 
 Using "guest mode" to run Trio on top of other event loops
diff --git a/pyproject.toml b/pyproject.toml
index f8d2f571e3..3121799cb7 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -51,6 +51,7 @@ module = [
     "trio._abc",
     "trio._core._asyncgens",
     "trio._core._entry_queue",
+    "trio._core._generated_run",
     "trio._core._generated_io_epoll",
     "trio._core._generated_io_kqueue",
     "trio._core._io_epoll",
@@ -58,6 +59,7 @@ module = [
     "trio._core._local",
     "trio._core._unbounded_queue",
     "trio._core._thread_cache",
+    "trio._core._run",
     "trio._deprecate",
     "trio._dtls",
     "trio._file_io",
@@ -73,7 +75,7 @@ disallow_untyped_defs = true
 disallow_untyped_decorators = true
 disallow_any_generics = true
 disallow_any_decorated = true
-disallow_any_unimported = true
+disallow_any_unimported = false  # Enable once outcome has stubs.
 disallow_subclassing_any = true
 
 [[tool.mypy.overrides]]
diff --git a/trio/__init__.py b/trio/__init__.py
index 277baa5339..be7de42cde 100644
--- a/trio/__init__.py
+++ b/trio/__init__.py
@@ -35,6 +35,7 @@
     EndOfChannel as EndOfChannel,
     Nursery as Nursery,
     RunFinishedError as RunFinishedError,
+    TaskStatus as TaskStatus,
     TrioInternalError as TrioInternalError,
     WouldBlock as WouldBlock,
     current_effective_deadline as current_effective_deadline,
diff --git a/trio/_core/__init__.py b/trio/_core/__init__.py
index 8e42d2743b..b9bd0d8cc4 100644
--- a/trio/_core/__init__.py
+++ b/trio/_core/__init__.py
@@ -27,6 +27,7 @@
     TASK_STATUS_IGNORED,
     CancelScope,
     Nursery,
+    RunStatistics,
     Task,
     TaskStatus,
     add_instrument,
diff --git a/trio/_core/_generated_run.py b/trio/_core/_generated_run.py
index 674c86aaec..35ecd45a1b 100644
--- a/trio/_core/_generated_run.py
+++ b/trio/_core/_generated_run.py
@@ -8,13 +8,21 @@
 
 from ._ki import LOCALS_KEY_KI_PROTECTION_ENABLED
 from ._run import GLOBAL_RUN_CONTEXT
-from ._run import _NO_SEND
+from collections.abc import Awaitable, Callable
+from typing import Any
 
+from outcome import Outcome
+import contextvars
 
-def current_statistics():
-    """Returns an object containing run-loop-level debugging information.
+from ._run import _NO_SEND, RunStatistics, Task
+from ._entry_queue import TrioToken
+from .._abc import Clock
 
-        Currently the following fields are defined:
+
+def current_statistics() ->RunStatistics:
+    """Returns ``RunStatistics``, which contains run-loop-level debugging information.
+
+        Currently, the following fields are defined:
 
         * ``tasks_living`` (int): The number of tasks that have been spawned
           and not yet exited.
@@ -41,7 +49,7 @@ def current_statistics():
         raise RuntimeError("must be called from async context")
 
 
-def current_time():
+def current_time() ->float:
     """Returns the current time according to Trio's internal clock.
 
         Returns:
@@ -58,7 +66,7 @@ def current_time():
         raise RuntimeError("must be called from async context")
 
 
-def current_clock():
+def current_clock() ->Clock:
     """Returns the current :class:`~trio.abc.Clock`."""
     locals()[LOCALS_KEY_KI_PROTECTION_ENABLED] = True
     try:
@@ -67,7 +75,7 @@ def current_clock():
         raise RuntimeError("must be called from async context")
 
 
-def current_root_task():
+def current_root_task() ->(Task | None):
     """Returns the current root :class:`Task`.
 
         This is the task that is the ultimate parent of all other tasks.
@@ -80,7 +88,7 @@ def current_root_task():
         raise RuntimeError("must be called from async context")
 
 
-def reschedule(task, next_send=_NO_SEND):
+def reschedule(task: Task, next_send: Outcome[Any]=_NO_SEND) ->None:  # type: ignore[has-type]
     """Reschedule the given task with the given
         :class:`outcome.Outcome`.
 
@@ -105,7 +113,9 @@ def reschedule(task, next_send=_NO_SEND):
         raise RuntimeError("must be called from async context")
 
 
-def spawn_system_task(async_fn, *args, name=None, context=None):
+def spawn_system_task(async_fn: Callable[..., Awaitable[object]], *args:
+    object, name: object=None, context: (contextvars.Context | None)=None
+    ) ->Task:
     """Spawn a "system" task.
 
         System tasks have a few differences from regular tasks:
@@ -164,7 +174,7 @@ def spawn_system_task(async_fn, *args, name=None, context=None):
         raise RuntimeError("must be called from async context")
 
 
-def current_trio_token():
+def current_trio_token() ->TrioToken:
     """Retrieve the :class:`TrioToken` for the current call to
         :func:`trio.run`.
 
@@ -176,7 +186,7 @@ def current_trio_token():
         raise RuntimeError("must be called from async context")
 
 
-async def wait_all_tasks_blocked(cushion=0.0):
+async def wait_all_tasks_blocked(cushion: float=0.0) ->None:
     """Block until there are no runnable tasks.
 
         This is useful in testing code when you want to give other tasks a
diff --git a/trio/_core/_io_epoll.py b/trio/_core/_io_epoll.py
index c4a31f3722..0d247cae64 100644
--- a/trio/_core/_io_epoll.py
+++ b/trio/_core/_io_epoll.py
@@ -15,6 +15,8 @@
 if TYPE_CHECKING:
     from socket import socket
 
+    from typing_extensions import TypeAlias
+
     from .._core import Abort, RaiseCancelT
 
 
@@ -28,6 +30,9 @@ class EpollWaiters:
 assert not TYPE_CHECKING or sys.platform == "linux"
 
 
+EventResult: TypeAlias = "list[tuple[int, int]]"
+
+
 @attr.s(slots=True, eq=False, frozen=True)
 class _EpollStatistics:
     tasks_waiting_read: int = attr.ib()
@@ -230,14 +235,14 @@ def force_wakeup(self) -> None:
     # Return value must be False-y IFF the timeout expired, NOT if any I/O
     # happened or force_wakeup was called. Otherwise it can be anything; gets
     # passed straight through to process_events.
-    def get_events(self, timeout: float) -> list[tuple[int, int]]:
+    def get_events(self, timeout: float) -> EventResult:
         # max_events must be > 0 or epoll gets cranky
         # accessing self._registered from a thread looks dangerous, but it's
         # OK because it doesn't matter if our value is a little bit off.
         max_events = max(1, len(self._registered))
         return self._epoll.poll(timeout, max_events)
 
-    def process_events(self, events: list[tuple[int, int]]) -> None:
+    def process_events(self, events: EventResult) -> None:
         for fd, flags in events:
             if fd == self._force_wakeup_fd:
                 self._force_wakeup.drain()
diff --git a/trio/_core/_io_kqueue.py b/trio/_core/_io_kqueue.py
index 0b0f8ee557..56a6559091 100644
--- a/trio/_core/_io_kqueue.py
+++ b/trio/_core/_io_kqueue.py
@@ -16,10 +16,14 @@
 if TYPE_CHECKING:
     from socket import socket
 
+    from typing_extensions import TypeAlias
+
     from .._core import Abort, RaiseCancelT, Task, UnboundedQueue
 
 assert not TYPE_CHECKING or (sys.platform != "linux" and sys.platform != "win32")
 
+EventResult: TypeAlias = "list[select.kevent]"
+
 
 @attr.s(slots=True, eq=False, frozen=True)
 class _KqueueStatistics:
@@ -62,7 +66,7 @@ def close(self) -> None:
     def force_wakeup(self) -> None:
         self._force_wakeup.wakeup_thread_and_signal_safe()
 
-    def get_events(self, timeout: float) -> list[select.kevent]:
+    def get_events(self, timeout: float) -> EventResult:
         # max_events must be > 0 or kqueue gets cranky
         # and we generally want this to be strictly larger than the actual
         # number of events we get, so that we can tell that we've gotten
@@ -79,7 +83,7 @@ def get_events(self, timeout: float) -> list[select.kevent]:
                 # and loop back to the start
         return events
 
-    def process_events(self, events: list[select.kevent]) -> None:
+    def process_events(self, events: EventResult) -> None:
         for event in events:
             key = (event.ident, event.filter)
             if event.ident == self._force_wakeup_fd:
diff --git a/trio/_core/_io_windows.py b/trio/_core/_io_windows.py
index 0130170af3..9757d25b5f 100644
--- a/trio/_core/_io_windows.py
+++ b/trio/_core/_io_windows.py
@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import enum
 import itertools
 import socket
@@ -29,6 +31,10 @@
 
 assert not TYPE_CHECKING or sys.platform == "win32"
 
+if TYPE_CHECKING:
+    from typing_extensions import TypeAlias
+EventResult: TypeAlias = int
+
 # There's a lot to be said about the overall design of a Windows event
 # loop. See
 #
@@ -365,10 +371,10 @@ class AFDGroup:
 
 @attr.s(slots=True, eq=False, frozen=True)
 class _WindowsStatistics:
-    tasks_waiting_read = attr.ib()
-    tasks_waiting_write = attr.ib()
-    tasks_waiting_overlapped = attr.ib()
-    completion_key_monitors = attr.ib()
+    tasks_waiting_read: int = attr.ib()
+    tasks_waiting_write: int = attr.ib()
+    tasks_waiting_overlapped: int = attr.ib()
+    completion_key_monitors: int = attr.ib()
     backend: Literal["windows"] = attr.ib(init=False, default="windows")
 
 
@@ -485,7 +491,7 @@ def force_wakeup(self):
             )
         )
 
-    def get_events(self, timeout):
+    def get_events(self, timeout: float) -> EventResult:
         received = ffi.new("PULONG")
         milliseconds = round(1000 * timeout)
         if timeout > 0 and milliseconds == 0:
@@ -500,9 +506,11 @@ def get_events(self, timeout):
             if exc.winerror != ErrorCodes.WAIT_TIMEOUT:  # pragma: no cover
                 raise
             return 0
-        return received[0]
+        result = received[0]
+        assert isinstance(result, int)
+        return result
 
-    def process_events(self, received):
+    def process_events(self, received: EventResult) -> None:
         for i in range(received):
             entry = self._events[i]
             if entry.lpCompletionKey == CKeys.AFD_POLL:
diff --git a/trio/_core/_local.py b/trio/_core/_local.py
index 39a2cff009..8286a5578f 100644
--- a/trio/_core/_local.py
+++ b/trio/_core/_local.py
@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Generic, TypeVar, final
+from typing import Generic, TypeVar, cast, final
 
 # Runvar implementations
 import attr
@@ -43,8 +43,7 @@ class RunVar(Generic[T], metaclass=Final):
     def get(self, default: T | type[_NoValue] = _NoValue) -> T:
         """Gets the value of this :class:`RunVar` for the current run call."""
         try:
-            # not typed yet
-            return _run.GLOBAL_RUN_CONTEXT.runner._locals[self]  # type: ignore[return-value, index]
+            return cast(T, _run.GLOBAL_RUN_CONTEXT.runner._locals[self])
         except AttributeError:
             raise RuntimeError("Cannot be used outside of a run context") from None
         except KeyError:
@@ -72,7 +71,7 @@ def set(self, value: T) -> RunVarToken[T]:
 
         # This can't fail, because if we weren't in Trio context then the
         # get() above would have failed.
-        _run.GLOBAL_RUN_CONTEXT.runner._locals[self] = value  # type: ignore[assignment, index]
+        _run.GLOBAL_RUN_CONTEXT.runner._locals[self] = value
         return token
 
     def reset(self, token: RunVarToken[T]) -> None:
@@ -92,9 +91,9 @@ def reset(self, token: RunVarToken[T]) -> None:
         previous = token.previous_value
         try:
             if previous is _NoValue:
-                _run.GLOBAL_RUN_CONTEXT.runner._locals.pop(self)  # type: ignore[arg-type]
+                _run.GLOBAL_RUN_CONTEXT.runner._locals.pop(self)
             else:
-                _run.GLOBAL_RUN_CONTEXT.runner._locals[self] = previous  # type: ignore[index, assignment]
+                _run.GLOBAL_RUN_CONTEXT.runner._locals[self] = previous
         except AttributeError:
             raise RuntimeError("Cannot be used outside of a run context")
 
diff --git a/trio/_core/_run.py b/trio/_core/_run.py
index 7d247a2738..1ba88da85e 100644
--- a/trio/_core/_run.py
+++ b/trio/_core/_run.py
@@ -10,14 +10,30 @@
 import threading
 import warnings
 from collections import deque
-from collections.abc import Callable, Coroutine, Iterator
+from collections.abc import (
+    Awaitable,
+    Callable,
+    Coroutine,
+    Generator,
+    Iterator,
+    Sequence,
+)
 from contextlib import AbstractAsyncContextManager, contextmanager
 from contextvars import copy_context
 from heapq import heapify, heappop, heappush
 from math import inf
 from time import perf_counter
 from types import TracebackType
-from typing import TYPE_CHECKING, Any, NoReturn, TypeVar
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    NoReturn,
+    Protocol,
+    TypeVar,
+    cast,
+    final,
+    overload,
+)
 
 import attr
 from outcome import Error, Outcome, Value, capture
@@ -25,6 +41,7 @@
 from sortedcontainers import SortedDict
 
 from .. import _core
+from .._abc import Clock, Instrument
 from .._util import Final, NoPublicConstructor, coroutine_or_error
 from ._asyncgens import AsyncGenerators
 from ._entry_queue import EntryQueue, TrioToken
@@ -53,11 +70,23 @@
     # An unfortunate name collision here with trio._util.Final
     from typing import Final as FinalT
 
+    from typing_extensions import Self
+
 DEADLINE_HEAP_MIN_PRUNE_THRESHOLD: FinalT = 1000
 
-_NO_SEND: FinalT = object()
+# Passed as a sentinel
+_NO_SEND: FinalT = cast("Outcome[Any]", object())
 
 FnT = TypeVar("FnT", bound="Callable[..., Any]")
+StatusT = TypeVar("StatusT")
+StatusT_co = TypeVar("StatusT_co", covariant=True)
+StatusT_contra = TypeVar("StatusT_contra", contravariant=True)
+RetT = TypeVar("RetT")
+
+
+@final
+class _NoStatus(metaclass=NoPublicConstructor):
+    """Sentinel for unset TaskStatus._value."""
 
 
 # Decorator to mark methods public. This does nothing by itself, but
@@ -120,7 +149,7 @@ def function_with_unique_name_xyzzy() -> NoReturn:
 
 
 @attr.s(frozen=True, slots=True)
-class SystemClock:
+class SystemClock(Clock):
     # Add a large random offset to our clock to ensure that if people
     # accidentally call time.perf_counter() directly or start comparing clocks
     # between different runs, then they'll notice the bug quickly:
@@ -149,7 +178,9 @@ class IdlePrimedTypes(enum.Enum):
 ################################################################
 
 
-def collapse_exception_group(excgroup):
+def collapse_exception_group(
+    excgroup: BaseExceptionGroup[BaseException],
+) -> BaseException:
     """Recursively collapse any single-exception groups into that single contained
     exception.
 
@@ -169,7 +200,8 @@ def collapse_exception_group(excgroup):
         )
         return exceptions[0]
     elif modified:
-        return excgroup.derive(exceptions)
+        # derive() returns Any for some reason.
+        return excgroup.derive(exceptions)  # type: ignore[no-any-return]
     else:
         return excgroup
 
@@ -184,18 +216,18 @@ class Deadlines:
     """
 
     # Heap of (deadline, id(CancelScope), CancelScope)
-    _heap = attr.ib(factory=list)
+    _heap: list[tuple[float, int, CancelScope]] = attr.ib(factory=list)
     # Count of active deadlines (those that haven't been changed)
-    _active = attr.ib(default=0)
+    _active: int = attr.ib(default=0)
 
-    def add(self, deadline, cancel_scope):
+    def add(self, deadline: float, cancel_scope: CancelScope) -> None:
         heappush(self._heap, (deadline, id(cancel_scope), cancel_scope))
         self._active += 1
 
-    def remove(self, deadline, cancel_scope):
+    def remove(self, deadline: float, cancel_scope: CancelScope) -> None:
         self._active -= 1
 
-    def next_deadline(self):
+    def next_deadline(self) -> float:
         while self._heap:
             deadline, _, cancel_scope = self._heap[0]
             if deadline == cancel_scope._registered_deadline:
@@ -205,7 +237,7 @@ def next_deadline(self):
                 heappop(self._heap)
         return inf
 
-    def _prune(self):
+    def _prune(self) -> None:
         # In principle, it's possible for a cancel scope to toggle back and
         # forth repeatedly between the same two deadlines, and end up with
         # lots of stale entries that *look* like they're still active, because
@@ -226,7 +258,7 @@ def _prune(self):
         heapify(pruned_heap)
         self._heap = pruned_heap
 
-    def expire(self, now):
+    def expire(self, now: float) -> bool:
         did_something = False
         while self._heap and self._heap[0][0] <= now:
             deadline, _, cancel_scope = heappop(self._heap)
@@ -377,14 +409,14 @@ def close(self) -> None:
                 child.recalculate()
 
     @property
-    def parent_cancellation_is_visible_to_us(self):
+    def parent_cancellation_is_visible_to_us(self) -> bool:
         return (
             self._parent is not None
             and not self._scope.shield
             and self._parent.effectively_cancelled
         )
 
-    def recalculate(self):
+    def recalculate(self) -> None:
         # This does a depth-first traversal over this and descendent cancel
         # statuses, to ensure their state is up-to-date. It's basically a
         # recursive algorithm, but we use an explicit stack to avoid any
@@ -403,7 +435,7 @@ def recalculate(self):
                         task._attempt_delivery_of_any_pending_cancel()
                 todo.extend(current._children)
 
-    def _mark_abandoned(self):
+    def _mark_abandoned(self) -> None:
         self.abandoned_by_misnesting = True
         for child in self._children:
             child._mark_abandoned()
@@ -440,6 +472,7 @@ def effective_deadline(self) -> float:
 """
 
 
+@final
 @attr.s(eq=False, repr=False, slots=True)
 class CancelScope(metaclass=Final):
     """A *cancellation scope*: the link between a unit of cancellable
@@ -491,7 +524,7 @@ class CancelScope(metaclass=Final):
     _shield: bool = attr.ib(default=False, kw_only=True)
 
     @enable_ki_protection
-    def __enter__(self):
+    def __enter__(self) -> Self:
         task = _core.current_task()
         if self._has_been_entered:
             raise RuntimeError(
@@ -505,7 +538,7 @@ def __enter__(self):
             task._activate_cancel_status(self._cancel_status)
         return self
 
-    def _close(self, exc):
+    def _close(self, exc: BaseException | None) -> BaseException | None:
         if self._cancel_status is None:
             new_exc = RuntimeError(
                 "Cancel scope stack corrupted: attempted to exit {!r} "
@@ -780,28 +813,57 @@ def cancel_called(self) -> bool:
 ################################################################
 
 
+class TaskStatus(Protocol[StatusT_contra]):
+    """The interface provided by :meth:`Nursery.start()` to the spawned task.
+
+    This is provided via the ``task_status`` keyword-only parameter.
+    """
+
+    @overload
+    def started(self: TaskStatus[None]) -> None:
+        ...
+
+    @overload
+    def started(self, value: StatusT_contra) -> None:
+        ...
+
+    def started(self, value: StatusT_contra | None = None) -> None:
+        """Tasks call this method to indicate that they have initialized.
+
+        See `nursery.start() <trio.Nursery.start>` for more information.
+        """
+
+
 # This code needs to be read alongside the code from Nursery.start to make
 # sense.
 @attr.s(eq=False, hash=False, repr=False)
-class TaskStatus(metaclass=Final):
-    _old_nursery = attr.ib()
-    _new_nursery = attr.ib()
-    _called_started = attr.ib(default=False)
-    _value = attr.ib(default=None)
+class _TaskStatus(TaskStatus[StatusT]):
+    _old_nursery: Nursery = attr.ib()
+    _new_nursery: Nursery = attr.ib()
+    # NoStatus is a sentinel.
+    _value: StatusT | type[_NoStatus] = attr.ib(default=_NoStatus)
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return f"<Task status object at {id(self):#x}>"
 
-    def started(self, value=None):
-        if self._called_started:
+    @overload
+    def started(self: _TaskStatus[None]) -> None:
+        ...
+
+    @overload
+    def started(self: _TaskStatus[StatusT], value: StatusT) -> None:
+        ...
+
+    def started(self, value: StatusT | None = None) -> None:
+        if self._value is not _NoStatus:
             raise RuntimeError("called 'started' twice on the same task status")
-        self._called_started = True
-        self._value = value
+        self._value = cast(StatusT, value)  # If None, StatusT == None
 
         # If the old nursery is cancelled, then quietly quit now; the child
         # will eventually exit on its own, and we don't want to risk moving
         # children that might have propagating Cancelled exceptions into
         # a place with no cancelled cancel scopes to catch them.
+        assert self._old_nursery._cancel_status is not None
         if self._old_nursery._cancel_status.effectively_cancelled:
             return
 
@@ -845,6 +907,15 @@ def started(self, value=None):
         self._old_nursery._check_nursery_closed()
 
 
+class _NurseryStartFunc(Protocol[StatusT_co]):
+    """Type of functions passed to `nursery.start() <trio.Nursery.start>`."""
+
+    def __call__(
+        self, *args: Any, task_status: TaskStatus[StatusT_co]
+    ) -> Awaitable[object]:
+        ...
+
+
 @attr.s
 class NurseryManager:
     """Nursery context manager.
@@ -934,6 +1005,7 @@ def open_nursery(
     return NurseryManager(strict_exception_groups=strict_exception_groups)
 
 
+@final
 class Nursery(metaclass=NoPublicConstructor):
     """A context which may be used to spawn (or cancel) child tasks.
 
@@ -997,20 +1069,22 @@ def _add_exc(self, exc: BaseException) -> None:
         self._pending_excs.append(exc)
         self.cancel_scope.cancel()
 
-    def _check_nursery_closed(self):
+    def _check_nursery_closed(self) -> None:
         if not any([self._nested_child_running, self._children, self._pending_starts]):
             self._closed = True
             if self._parent_waiting_in_aexit:
                 self._parent_waiting_in_aexit = False
                 GLOBAL_RUN_CONTEXT.runner.reschedule(self._parent_task)
 
-    def _child_finished(self, task, outcome):
+    def _child_finished(self, task: Task, outcome: Outcome[Any]) -> None:
         self._children.remove(task)
         if isinstance(outcome, Error):
             self._add_exc(outcome.error)
         self._check_nursery_closed()
 
-    async def _nested_child_finished(self, nested_child_exc):
+    async def _nested_child_finished(
+        self, nested_child_exc: BaseException | None
+    ) -> BaseException | None:
         # Returns MultiError instance (or any exception if the nursery is in loose mode
         # and there is just one contained exception) if there are pending exceptions
         if nested_child_exc is not None:
@@ -1022,7 +1096,7 @@ async def _nested_child_finished(self, nested_child_exc):
             # If we get cancelled (or have an exception injected, like
             # KeyboardInterrupt), then save that, but still wait until our
             # children finish.
-            def aborted(raise_cancel):
+            def aborted(raise_cancel: _core.RaiseCancelT) -> Abort:
                 self._add_exc(capture(raise_cancel).error)
                 return Abort.FAILED
 
@@ -1048,8 +1122,15 @@ def aborted(raise_cancel):
                 # avoid a garbage cycle
                 # (see test_nursery_cancel_doesnt_create_cyclic_garbage)
                 del self._pending_excs
+        return None
 
-    def start_soon(self, async_fn, *args, name=None):
+    def start_soon(
+        self,
+        # TODO: TypeVarTuple
+        async_fn: Callable[..., Awaitable[object]],
+        *args: object,
+        name: object = None,
+    ) -> None:
         """Creates a child task, scheduling ``await async_fn(*args)``.
 
         If you want to run a function and immediately wait for its result,
@@ -1091,7 +1172,9 @@ def start_soon(self, async_fn, *args, name=None):
         """
         GLOBAL_RUN_CONTEXT.runner.spawn_impl(async_fn, args, self, name)
 
-    async def start(self, async_fn, *args, name=None):
+    async def start(
+        self, async_fn: _NurseryStartFunc[StatusT], *args: object, name: object = None
+    ) -> StatusT:
         r"""Creates and initializes a child task.
 
         Like :meth:`start_soon`, but blocks until the new task has
@@ -1100,7 +1183,7 @@ async def start(self, async_fn, *args, name=None):
 
         The ``async_fn`` must accept a ``task_status`` keyword argument,
         and it must make sure that it (or someone) eventually calls
-        ``task_status.started()``.
+        :meth:`task_status.started() <TaskStatus.started>`.
 
         The conventional way to define ``async_fn`` is like::
 
@@ -1113,31 +1196,30 @@ async def async_fn(arg1, arg2, *, task_status=trio.TASK_STATUS_IGNORED):
         a do-nothing ``started`` method. This way your function supports
         being called either like ``await nursery.start(async_fn, arg1,
         arg2)`` or directly like ``await async_fn(arg1, arg2)``, and
-        either way it can call ``task_status.started()`` without
-        worrying about which mode it's in. Defining your function like
+        either way it can call :meth:`task_status.started() <TaskStatus.started>`
+        without worrying about which mode it's in. Defining your function like
         this will make it obvious to readers that it supports being used
         in both modes.
 
-        Before the child calls ``task_status.started()``, it's
-        effectively run underneath the call to :meth:`start`: if it
+        Before the child calls :meth:`task_status.started() <TaskStatus.started>`,
+        it's effectively run underneath the call to :meth:`start`: if it
         raises an exception then that exception is reported by
         :meth:`start`, and does *not* propagate out of the nursery. If
         :meth:`start` is cancelled, then the child task is also
         cancelled.
 
-        When the child calls ``task_status.started()``, it's moved out
-        from underneath :meth:`start` and into the given nursery.
+        When the child calls :meth:`task_status.started() <TaskStatus.started>`,
+        it's moved out from underneath :meth:`start` and into the given nursery.
 
-        If the child task passes a value to
-        ``task_status.started(value)``, then :meth:`start` returns this
-        value. Otherwise it returns ``None``.
+        If the child task passes a value to :meth:`task_status.started(value) <TaskStatus.started>`,
+        then :meth:`start` returns this value. Otherwise, it returns ``None``.
         """
         if self._closed:
             raise RuntimeError("Nursery is closed to new arrivals")
         try:
             self._pending_starts += 1
             async with open_nursery() as old_nursery:
-                task_status = TaskStatus(old_nursery, self)
+                task_status: _TaskStatus[StatusT] = _TaskStatus(old_nursery, self)
                 thunk = functools.partial(async_fn, task_status=task_status)
                 task = GLOBAL_RUN_CONTEXT.runner.spawn_impl(
                     thunk, args, old_nursery, name
@@ -1148,9 +1230,9 @@ async def async_fn(arg1, arg2, *, task_status=trio.TASK_STATUS_IGNORED):
             # If we get here, then the child either got reparented or exited
             # normally. The complicated logic is all in TaskStatus.started().
             # (Any exceptions propagate directly out of the above.)
-            if not task_status._called_started:
+            if task_status._value is _NoStatus:
                 raise RuntimeError("child exited without calling task_status.started()")
-            return task_status._value
+            return task_status._value  # type: ignore[return-value]  # Mypy doesn't narrow yet.
         finally:
             self._pending_starts -= 1
             self._check_nursery_closed()
@@ -1164,11 +1246,12 @@ def __del__(self) -> None:
 ################################################################
 
 
+@final
 @attr.s(eq=False, hash=False, repr=False, slots=True)
 class Task(metaclass=NoPublicConstructor):
     _parent_nursery: Nursery | None = attr.ib()
     coro: Coroutine[Any, Outcome[object], Any] = attr.ib()
-    _runner = attr.ib()
+    _runner: Runner = attr.ib()
     name: str = attr.ib()
     context: contextvars.Context = attr.ib()
     _counter: int = attr.ib(init=False, factory=itertools.count().__next__)
@@ -1184,11 +1267,9 @@ class Task(metaclass=NoPublicConstructor):
     #   tracebacks with extraneous frames.
     # - for scheduled tasks, custom_sleep_data is None
     # Tasks start out unscheduled.
-    _next_send_fn = attr.ib(default=None)
-    _next_send = attr.ib(default=None)
-    _abort_func: Callable[[Callable[[], NoReturn]], Abort] | None = attr.ib(
-        default=None
-    )
+    _next_send_fn: Callable[[Any], object] = attr.ib(default=None)
+    _next_send: Outcome[Any] | None | BaseException = attr.ib(default=None)
+    _abort_func: Callable[[_core.RaiseCancelT], Abort] | None = attr.ib(default=None)
     custom_sleep_data: Any = attr.ib(default=None)
 
     # For introspection and nursery.start()
@@ -1256,7 +1337,7 @@ def print_stack_for_task(task):
                 print("".join(ss.format()))
 
         """
-        # ignore static typing as we're doing lots of dynamic introspection
+        # Ignore static typing as we're doing lots of dynamic introspection
         coro: Any = self.coro
         while coro is not None:
             if hasattr(coro, "cr_frame"):
@@ -1290,18 +1371,19 @@ def print_stack_for_task(task):
 
     # The CancelStatus object that is currently active for this task.
     # Don't change this directly; instead, use _activate_cancel_status().
+    # This can be None, but only in the init task.
     _cancel_status: CancelStatus = attr.ib(default=None, repr=False)
 
-    def _activate_cancel_status(self, cancel_status: CancelStatus) -> None:
+    def _activate_cancel_status(self, cancel_status: CancelStatus | None) -> None:
         if self._cancel_status is not None:
             self._cancel_status._tasks.remove(self)
-        self._cancel_status = cancel_status
+        self._cancel_status = cancel_status  # type: ignore[assignment]
         if self._cancel_status is not None:
             self._cancel_status._tasks.add(self)
             if self._cancel_status.effectively_cancelled:
                 self._attempt_delivery_of_any_pending_cancel()
 
-    def _attempt_abort(self, raise_cancel: Callable[[], NoReturn]) -> None:
+    def _attempt_abort(self, raise_cancel: _core.RaiseCancelT) -> None:
         # Either the abort succeeds, in which case we will reschedule the
         # task, or else it fails, in which case it will worry about
         # rescheduling itself (hopefully eventually calling reraise to raise
@@ -1326,7 +1408,7 @@ def _attempt_delivery_of_any_pending_cancel(self) -> None:
         if not self._cancel_status.effectively_cancelled:
             return
 
-        def raise_cancel():
+        def raise_cancel() -> NoReturn:
             raise Cancelled._create()
 
         self._attempt_abort(raise_cancel)
@@ -1356,13 +1438,35 @@ class RunContext(threading.local):
 GLOBAL_RUN_CONTEXT: FinalT = RunContext()
 
 
-@attr.s(frozen=True)
-class _RunStatistics:
-    tasks_living = attr.ib()
-    tasks_runnable = attr.ib()
-    seconds_to_next_deadline = attr.ib()
-    io_statistics = attr.ib()
-    run_sync_soon_queue_size = attr.ib()
+@attr.frozen
+class RunStatistics:
+    """An object containing run-loop-level debugging information.
+
+    Currently, the following fields are defined:
+
+    * ``tasks_living`` (int): The number of tasks that have been spawned
+      and not yet exited.
+    * ``tasks_runnable`` (int): The number of tasks that are currently
+      queued on the run queue (as opposed to blocked waiting for something
+      to happen).
+    * ``seconds_to_next_deadline`` (float): The time until the next
+      pending cancel scope deadline. May be negative if the deadline has
+      expired but we haven't yet processed cancellations. May be
+      :data:`~math.inf` if there are no pending deadlines.
+    * ``run_sync_soon_queue_size`` (int): The number of
+      unprocessed callbacks queued via
+      :meth:`trio.lowlevel.TrioToken.run_sync_soon`.
+    * ``io_statistics`` (object): Some statistics from Trio's I/O
+      backend. This always has an attribute ``backend`` which is a string
+      naming which operating-system-specific I/O backend is in use; the
+      other attributes vary between backends.
+    """
+
+    tasks_living: int
+    tasks_runnable: int
+    seconds_to_next_deadline: float
+    io_statistics: IOStatistics
+    run_sync_soon_queue_size: int
 
 
 # This holds all the state that gets trampolined back and forth between
@@ -1386,18 +1490,19 @@ class _RunStatistics:
 # worker thread.
 @attr.s(eq=False, hash=False, slots=True)
 class GuestState:
-    runner = attr.ib()
-    run_sync_soon_threadsafe = attr.ib()
-    run_sync_soon_not_threadsafe = attr.ib()
-    done_callback = attr.ib()
-    unrolled_run_gen = attr.ib()
-    _value_factory: Callable[[], Value] = lambda: Value(None)
-    unrolled_run_next_send = attr.ib(factory=_value_factory, type=Outcome)
-
-    def guest_tick(self):
+    runner: Runner = attr.ib()
+    run_sync_soon_threadsafe: Callable[[Callable[[], object]], object] = attr.ib()
+    run_sync_soon_not_threadsafe: Callable[[Callable[[], object]], object] = attr.ib()
+    done_callback: Callable[[Outcome[Any]], object] = attr.ib()
+    unrolled_run_gen: Generator[float, EventResult, None] = attr.ib()
+    _value_factory: Callable[[], Value[Any]] = lambda: Value(None)
+    unrolled_run_next_send: Outcome[Any] = attr.ib(factory=_value_factory)
+
+    def guest_tick(self) -> None:
         try:
             timeout = self.unrolled_run_next_send.send(self.unrolled_run_gen)
         except StopIteration:
+            assert self.runner.main_task_outcome is not None
             self.done_callback(self.runner.main_task_outcome)
             return
         except TrioInternalError as exc:
@@ -1405,7 +1510,9 @@ def guest_tick(self):
             return
 
         # Optimization: try to skip going into the thread if we can avoid it
-        events_outcome = capture(self.runner.io_manager.get_events, 0)
+        events_outcome: Value[EventResult] | Error = capture(
+            self.runner.io_manager.get_events, 0
+        )
         if timeout <= 0 or isinstance(events_outcome, Error) or events_outcome.value:
             # No need to go into the thread
             self.unrolled_run_next_send = events_outcome
@@ -1415,11 +1522,11 @@ def guest_tick(self):
             # Need to go into the thread and call get_events() there
             self.runner.guest_tick_scheduled = False
 
-            def get_events():
+            def get_events() -> EventResult:
                 return self.runner.io_manager.get_events(timeout)
 
-            def deliver(events_outcome):
-                def in_main_thread():
+            def deliver(events_outcome: Outcome[EventResult]) -> None:
+                def in_main_thread() -> None:
                     self.unrolled_run_next_send = events_outcome
                     self.runner.guest_tick_scheduled = True
                     self.guest_tick()
@@ -1431,44 +1538,44 @@ def in_main_thread():
 
 @attr.s(eq=False, hash=False, slots=True)
 class Runner:
-    clock = attr.ib()
+    clock: Clock = attr.ib()
     instruments: Instruments = attr.ib()
     io_manager: TheIOManager = attr.ib()
-    ki_manager = attr.ib()
-    strict_exception_groups = attr.ib()
+    ki_manager: KIManager = attr.ib()
+    strict_exception_groups: bool = attr.ib()
 
     # Run-local values, see _local.py
-    _locals = attr.ib(factory=dict)
+    _locals: dict[_core.RunVar[Any], Any] = attr.ib(factory=dict)
 
     runq: deque[Task] = attr.ib(factory=deque)
-    tasks = attr.ib(factory=set)
+    tasks: set[Task] = attr.ib(factory=set)
 
-    deadlines = attr.ib(factory=Deadlines)
+    deadlines: Deadlines = attr.ib(factory=Deadlines)
 
-    init_task = attr.ib(default=None)
-    system_nursery = attr.ib(default=None)
-    system_context = attr.ib(default=None)
-    main_task = attr.ib(default=None)
-    main_task_outcome = attr.ib(default=None)
+    init_task: Task | None = attr.ib(default=None)
+    system_nursery: Nursery | None = attr.ib(default=None)
+    system_context: contextvars.Context = attr.ib(kw_only=True)
+    main_task: Task | None = attr.ib(default=None)
+    main_task_outcome: Outcome[Any] | None = attr.ib(default=None)
 
-    entry_queue = attr.ib(factory=EntryQueue)
-    trio_token = attr.ib(default=None)
-    asyncgens = attr.ib(factory=AsyncGenerators)
+    entry_queue: EntryQueue = attr.ib(factory=EntryQueue)
+    trio_token: TrioToken | None = attr.ib(default=None)
+    asyncgens: AsyncGenerators = attr.ib(factory=AsyncGenerators)
 
     # If everything goes idle for this long, we call clock._autojump()
-    clock_autojump_threshold = attr.ib(default=inf)
+    clock_autojump_threshold: float = attr.ib(default=inf)
 
     # Guest mode stuff
-    is_guest = attr.ib(default=False)
-    guest_tick_scheduled = attr.ib(default=False)
+    is_guest: bool = attr.ib(default=False)
+    guest_tick_scheduled: bool = attr.ib(default=False)
 
-    def force_guest_tick_asap(self):
+    def force_guest_tick_asap(self) -> None:
         if self.guest_tick_scheduled:
             return
         self.guest_tick_scheduled = True
         self.io_manager.force_wakeup()
 
-    def close(self):
+    def close(self) -> None:
         self.io_manager.close()
         self.entry_queue.close()
         self.asyncgens.close()
@@ -1478,10 +1585,10 @@ def close(self):
         self.ki_manager.close()
 
     @_public
-    def current_statistics(self):
-        """Returns an object containing run-loop-level debugging information.
+    def current_statistics(self) -> RunStatistics:
+        """Returns ``RunStatistics``, which contains run-loop-level debugging information.
 
-        Currently the following fields are defined:
+        Currently, the following fields are defined:
 
         * ``tasks_living`` (int): The number of tasks that have been spawned
           and not yet exited.
@@ -1502,7 +1609,7 @@ def current_statistics(self):
 
         """
         seconds_to_next_deadline = self.deadlines.next_deadline() - self.current_time()
-        return _RunStatistics(
+        return RunStatistics(
             tasks_living=len(self.tasks),
             tasks_runnable=len(self.runq),
             seconds_to_next_deadline=seconds_to_next_deadline,
@@ -1511,7 +1618,7 @@ def current_statistics(self):
         )
 
     @_public
-    def current_time(self):
+    def current_time(self) -> float:
         """Returns the current time according to Trio's internal clock.
 
         Returns:
@@ -1524,12 +1631,12 @@ def current_time(self):
         return self.clock.current_time()
 
     @_public
-    def current_clock(self):
+    def current_clock(self) -> Clock:
         """Returns the current :class:`~trio.abc.Clock`."""
         return self.clock
 
     @_public
-    def current_root_task(self):
+    def current_root_task(self) -> Task | None:
         """Returns the current root :class:`Task`.
 
         This is the task that is the ultimate parent of all other tasks.
@@ -1541,8 +1648,10 @@ def current_root_task(self):
     # Core task handling primitives
     ################
 
-    @_public
-    def reschedule(self, task, next_send=_NO_SEND):
+    @_public  # Type-ignore due to use of Any here.
+    def reschedule(  # type: ignore[misc]
+        self, task: Task, next_send: Outcome[Any] = _NO_SEND
+    ) -> None:
         """Reschedule the given task with the given
         :class:`outcome.Outcome`.
 
@@ -1576,8 +1685,16 @@ def reschedule(self, task, next_send=_NO_SEND):
             self.instruments.call("task_scheduled", task)
 
     def spawn_impl(
-        self, async_fn, args, nursery, name, *, system_task=False, context=None
-    ):
+        self,
+        # TODO: TypeVarTuple
+        async_fn: Callable[..., Awaitable[object]],
+        args: tuple[object, ...],
+        nursery: Nursery | None,
+        name: object,
+        *,
+        system_task: bool = False,
+        context: contextvars.Context | None = None,
+    ) -> Task:
         ######
         # Make sure the nursery is in working order
         ######
@@ -1607,7 +1724,8 @@ def spawn_impl(
         # Call the function and get the coroutine object, while giving helpful
         # errors for common mistakes.
         ######
-        coro = context.run(coroutine_or_error, async_fn, *args)
+        # TODO: resolve the type: ignore when implementing TypeVarTuple
+        coro = context.run(coroutine_or_error, async_fn, *args)  # type: ignore[arg-type]
 
         if name is None:
             name = async_fn
@@ -1615,13 +1733,13 @@ def spawn_impl(
             name = name.func
         if not isinstance(name, str):
             try:
-                name = f"{name.__module__}.{name.__qualname__}"
+                name = f"{name.__module__}.{name.__qualname__}"  # type: ignore[attr-defined]
             except AttributeError:
                 name = repr(name)
 
         if not hasattr(coro, "cr_frame"):
             # This async function is implemented in C or Cython
-            async def python_wrapper(orig_coro):
+            async def python_wrapper(orig_coro: Awaitable[RetT]) -> RetT:
                 return await orig_coro
 
             coro = python_wrapper(coro)
@@ -1643,10 +1761,11 @@ async def python_wrapper(orig_coro):
             self.instruments.call("task_spawned", task)
         # Special case: normally next_send should be an Outcome, but for the
         # very first send we have to send a literal unboxed None.
-        self.reschedule(task, None)
+        # TODO: remove [unused-ignore] when Outcome is typed
+        self.reschedule(task, None)  # type: ignore[arg-type, unused-ignore]
         return task
 
-    def task_exited(self, task, outcome):
+    def task_exited(self, task: Task, outcome: Outcome[Any]) -> None:
         if (
             task._cancel_status is not None
             and task._cancel_status.abandoned_by_misnesting
@@ -1685,6 +1804,7 @@ def task_exited(self, task, outcome):
             if task is self.main_task:
                 self.main_task_outcome = outcome
                 outcome = Value(None)
+            assert task._parent_nursery is not None, task
             task._parent_nursery._child_finished(task, outcome)
 
         if "task_exited" in self.instruments:
@@ -1694,8 +1814,15 @@ def task_exited(self, task, outcome):
     # System tasks and init
     ################
 
-    @_public
-    def spawn_system_task(self, async_fn, *args, name=None, context=None):
+    @_public  # Type-ignore due to use of Any here.
+    def spawn_system_task(  # type: ignore[misc]
+        self,
+        # TODO: TypeVarTuple
+        async_fn: Callable[..., Awaitable[object]],
+        *args: object,
+        name: object = None,
+        context: contextvars.Context | None = None,
+    ) -> Task:
         """Spawn a "system" task.
 
         System tasks have a few differences from regular tasks:
@@ -1756,7 +1883,12 @@ def spawn_system_task(self, async_fn, *args, name=None, context=None):
             context=context,
         )
 
-    async def init(self, async_fn, args):
+    async def init(
+        # TODO: TypeVarTuple
+        self,
+        async_fn: Callable[..., Awaitable[object]],
+        args: tuple[object, ...],
+    ) -> None:
         # run_sync_soon task runs here:
         async with open_nursery() as run_sync_soon_nursery:
             # All other system tasks run here:
@@ -1794,7 +1926,7 @@ async def init(self, async_fn, args):
     ################
 
     @_public
-    def current_trio_token(self):
+    def current_trio_token(self) -> TrioToken:
         """Retrieve the :class:`TrioToken` for the current call to
         :func:`trio.run`.
 
@@ -1807,7 +1939,7 @@ def current_trio_token(self):
     # KI handling
     ################
 
-    ki_pending = attr.ib(default=False)
+    ki_pending: bool = attr.ib(default=False)
 
     # deliver_ki is broke. Maybe move all the actual logic and state into
     # RunToken, and we'll only have one instance per runner? But then we can't
@@ -1816,14 +1948,14 @@ def current_trio_token(self):
     # keep the class public so people can isinstance() it if they want.
 
     # This gets called from signal context
-    def deliver_ki(self):
+    def deliver_ki(self) -> None:
         self.ki_pending = True
         try:
             self.entry_queue.run_sync_soon(self._deliver_ki_cb)
         except RunFinishedError:
             pass
 
-    def _deliver_ki_cb(self):
+    def _deliver_ki_cb(self) -> None:
         if not self.ki_pending:
             return
         # Can't happen because main_task and run_sync_soon_task are created at
@@ -1840,10 +1972,12 @@ def _deliver_ki_cb(self):
     # Quiescing
     ################
 
-    waiting_for_idle = attr.ib(factory=SortedDict)
+    # sortedcontainers doesn't have types, and is reportedly very hard to type:
+    # https://github.com/grantjenks/python-sortedcontainers/issues/68
+    waiting_for_idle: Any = attr.ib(factory=SortedDict)
 
     @_public
-    async def wait_all_tasks_blocked(self, cushion=0.0):
+    async def wait_all_tasks_blocked(self, cushion: float = 0.0) -> None:
         """Block until there are no runnable tasks.
 
         This is useful in testing code when you want to give other tasks a
@@ -1905,7 +2039,7 @@ async def test_lock_fairness():
         key = (cushion, id(task))
         self.waiting_for_idle[key] = task
 
-        def abort(_):
+        def abort(_: _core.RaiseCancelT) -> Abort:
             del self.waiting_for_idle[key]
             return Abort.SUCCEEDED
 
@@ -1980,11 +2114,11 @@ def abort(_):
 
 
 def setup_runner(
-    clock,
-    instruments,
-    restrict_keyboard_interrupt_to_checkpoints,
-    strict_exception_groups,
-):
+    clock: Clock | None,
+    instruments: Sequence[Instrument],
+    restrict_keyboard_interrupt_to_checkpoints: bool,
+    strict_exception_groups: bool,
+) -> Runner:
     """Create a Runner object and install it as the GLOBAL_RUN_CONTEXT."""
     # It wouldn't be *hard* to support nested calls to run(), but I can't
     # think of a single good reason for it, so let's be conservative for
@@ -1994,14 +2128,14 @@ def setup_runner(
 
     if clock is None:
         clock = SystemClock()
-    instruments = Instruments(instruments)
+    instrument_group = Instruments(instruments)
     io_manager = TheIOManager()
     system_context = copy_context()
     ki_manager = KIManager()
 
     runner = Runner(
         clock=clock,
-        instruments=instruments,
+        instruments=instrument_group,
         io_manager=io_manager,
         system_context=system_context,
         ki_manager=ki_manager,
@@ -2018,13 +2152,13 @@ def setup_runner(
 
 
 def run(
-    async_fn,
-    *args,
-    clock=None,
-    instruments=(),
+    async_fn: Callable[..., RetT],
+    *args: object,
+    clock: Clock | None = None,
+    instruments: Sequence[Instrument] = (),
     restrict_keyboard_interrupt_to_checkpoints: bool = False,
     strict_exception_groups: bool = False,
-):
+) -> RetT:
     """Run a Trio-flavored async function, and return the result.
 
     Calling::
@@ -2107,7 +2241,8 @@ def run(
     )
 
     gen = unrolled_run(runner, async_fn, args)
-    next_send = None
+    # Need to send None in the first time.
+    next_send: EventResult = None  # type: ignore[assignment]
     while True:
         try:
             timeout = gen.send(next_send)
@@ -2117,23 +2252,26 @@ def run(
     # Inlined copy of runner.main_task_outcome.unwrap() to avoid
     # cluttering every single Trio traceback with an extra frame.
     if isinstance(runner.main_task_outcome, Value):
-        return runner.main_task_outcome.value
-    else:
+        return cast(RetT, runner.main_task_outcome.value)
+    elif isinstance(runner.main_task_outcome, Error):
         raise runner.main_task_outcome.error
+    else:  # pragma: no cover
+        raise AssertionError(runner.main_task_outcome)
 
 
 def start_guest_run(
-    async_fn,
-    *args,
-    run_sync_soon_threadsafe,
-    done_callback,
-    run_sync_soon_not_threadsafe=None,
+    async_fn: Callable[..., Awaitable[RetT]],
+    *args: object,
+    run_sync_soon_threadsafe: Callable[[Callable[[], object]], object],
+    done_callback: Callable[[Outcome[RetT]], object],
+    run_sync_soon_not_threadsafe: Callable[[Callable[[], object]], object]
+    | None = None,
     host_uses_signal_set_wakeup_fd: bool = False,
-    clock=None,
-    instruments=(),
+    clock: Clock | None = None,
+    instruments: Sequence[Instrument] = (),
     restrict_keyboard_interrupt_to_checkpoints: bool = False,
     strict_exception_groups: bool = False,
-):
+) -> None:
     """Start a "guest" run of Trio on top of some other "host" event loop.
 
     Each host loop can only have one guest run at a time.
@@ -2221,10 +2359,10 @@ def my_done_callback(run_outcome):
 # straight through.
 def unrolled_run(
     runner: Runner,
-    async_fn,
-    args,
+    async_fn: Callable[..., object],
+    args: tuple[object, ...],
     host_uses_signal_set_wakeup_fd: bool = False,
-):
+) -> Generator[float, EventResult, None]:
     locals()[LOCALS_KEY_KI_PROTECTION_ENABLED] = True
     __tracebackhide__ = True
 
@@ -2310,6 +2448,7 @@ def unrolled_run(
                             break
                 else:
                     assert idle_primed is IdlePrimedTypes.AUTOJUMP_CLOCK
+                    assert isinstance(runner.clock, _core.MockClock)
                     runner.clock._autojump()
 
             # Process all runnable tasks, but only the ones that are already
@@ -2354,7 +2493,7 @@ def unrolled_run(
                 next_send_fn = task._next_send_fn
                 next_send = task._next_send
                 task._next_send_fn = task._next_send = None
-                final_outcome = None
+                final_outcome: Outcome[Any] | None = None
                 try:
                     # We used to unwrap the Outcome object here and send/throw
                     # its contents in directly, but it turns out that .throw()
@@ -2373,9 +2512,8 @@ def unrolled_run(
                     # more Context.run adds.
                     tb = task_exc.__traceback__
                     for _ in range(1 + CONTEXT_RUN_TB_FRAMES):
-                        if tb is None:
-                            break
-                        tb = tb.tb_next
+                        if tb is not None:  # pragma: no branch
+                            tb = tb.tb_next
                     final_outcome = Error(task_exc.with_traceback(tb))
                     # Remove local refs so that e.g. cancelled coroutine locals
                     # are not kept alive by this frame until another exception
@@ -2418,7 +2556,8 @@ def unrolled_run(
                         # protocol of unwrapping whatever outcome gets sent in.
                         # Instead, we'll arrange to throw `exc` in directly,
                         # which works for at least asyncio and curio.
-                        runner.reschedule(task, exc)
+                        # TODO: remove [unused-ignore] when Outcome is typed
+                        runner.reschedule(task, exc)  # type: ignore[arg-type, unused-ignore]
                         task._next_send_fn = task.coro.throw
                     # prevent long-lived reference
                     # TODO: develop test for this deletion
@@ -2461,15 +2600,15 @@ def unrolled_run(
 ################################################################
 
 
-class _TaskStatusIgnored:
+class _TaskStatusIgnored(TaskStatus[Any]):
     def __repr__(self) -> str:
         return "TASK_STATUS_IGNORED"
 
-    def started(self, value: object = None) -> None:
+    def started(self, value: Any = None) -> None:
         pass
 
 
-TASK_STATUS_IGNORED: FinalT = _TaskStatusIgnored()
+TASK_STATUS_IGNORED: FinalT[TaskStatus[Any]] = _TaskStatusIgnored()
 
 
 def current_task() -> Task:
@@ -2566,13 +2705,25 @@ async def checkpoint_if_cancelled() -> None:
 
 if sys.platform == "win32":
     from ._generated_io_windows import *
-    from ._io_windows import WindowsIOManager as TheIOManager
+    from ._io_windows import (
+        EventResult as EventResult,
+        WindowsIOManager as TheIOManager,
+        _WindowsStatistics as IOStatistics,
+    )
 elif sys.platform == "linux" or (not TYPE_CHECKING and hasattr(select, "epoll")):
     from ._generated_io_epoll import *
-    from ._io_epoll import EpollIOManager as TheIOManager
+    from ._io_epoll import (
+        EpollIOManager as TheIOManager,
+        EventResult as EventResult,
+        _EpollStatistics as IOStatistics,
+    )
 elif TYPE_CHECKING or hasattr(select, "kqueue"):
     from ._generated_io_kqueue import *
-    from ._io_kqueue import KqueueIOManager as TheIOManager
+    from ._io_kqueue import (
+        EventResult as EventResult,
+        KqueueIOManager as TheIOManager,
+        _KqueueStatistics as IOStatistics,
+    )
 else:  # pragma: no cover
     raise NotImplementedError("unsupported platform")
 
diff --git a/trio/_core/_thread_cache.py b/trio/_core/_thread_cache.py
index d8f5a94c5b..8381153576 100644
--- a/trio/_core/_thread_cache.py
+++ b/trio/_core/_thread_cache.py
@@ -118,7 +118,7 @@ def darwin_namefunc(
 
 class WorkerThread(Generic[RetT]):
     def __init__(self, thread_cache: ThreadCache) -> None:
-        self._job: tuple[  # type: ignore[no-any-unimported]
+        self._job: tuple[
             Callable[[], RetT],
             Callable[[outcome.Outcome[RetT]], object],
             str | None,
@@ -198,7 +198,7 @@ class ThreadCache:
     def __init__(self) -> None:
         self._idle_workers: dict[WorkerThread[Any], None] = {}
 
-    def start_thread_soon(  # type: ignore[no-any-unimported]
+    def start_thread_soon(
         self,
         fn: Callable[[], RetT],
         deliver: Callable[[outcome.Outcome[RetT]], object],
@@ -216,7 +216,7 @@ def start_thread_soon(  # type: ignore[no-any-unimported]
 THREAD_CACHE = ThreadCache()
 
 
-def start_thread_soon(  # type: ignore[no-any-unimported]
+def start_thread_soon(
     fn: Callable[[], RetT],
     deliver: Callable[[outcome.Outcome[RetT]], object],
     name: str | None = None,
diff --git a/trio/_dtls.py b/trio/_dtls.py
index b3ed0fd883..08b7672a2f 100644
--- a/trio/_dtls.py
+++ b/trio/_dtls.py
@@ -42,7 +42,6 @@
     from OpenSSL.SSL import Context
     from typing_extensions import Self, TypeAlias
 
-    from trio.lowlevel import TaskStatus
     from trio.socket import Address, _SocketType
 
 MAX_UDP_PACKET_SIZE = 65527
@@ -1267,7 +1266,7 @@ async def serve(
         ssl_context: Context,
         async_fn: Callable[..., Awaitable[object]],
         *args: Any,
-        task_status: TaskStatus = trio.TASK_STATUS_IGNORED,  # type: ignore[has-type]
+        task_status: trio.TaskStatus[None] = trio.TASK_STATUS_IGNORED,
     ) -> None:
         """Listen for incoming connections, and spawn a handler for each using an
         internal nursery.
diff --git a/trio/_highlevel_open_tcp_listeners.py b/trio/_highlevel_open_tcp_listeners.py
index 0d5f630495..e6840eae97 100644
--- a/trio/_highlevel_open_tcp_listeners.py
+++ b/trio/_highlevel_open_tcp_listeners.py
@@ -6,7 +6,7 @@
 from math import inf
 
 import trio
-from trio.lowlevel import TaskStatus
+from trio import TaskStatus
 
 from . import socket as tsocket
 
@@ -171,7 +171,7 @@ async def serve_tcp(
     host: str | bytes | None = None,
     backlog: int | float | None = None,
     handler_nursery: trio.Nursery | None = None,
-    task_status: TaskStatus = trio.TASK_STATUS_IGNORED,  # type: ignore[assignment]  # default has type "_TaskStatusIgnored", argument has type "TaskStatus"
+    task_status: TaskStatus[list[trio.SocketListener]] = trio.TASK_STATUS_IGNORED,
 ) -> None:
     """Listen for incoming TCP connections, and for each one start a task
     running ``handler(stream)``.
diff --git a/trio/_tests/test_exports.py b/trio/_tests/test_exports.py
index 1b1e8df8da..c0da975fb4 100644
--- a/trio/_tests/test_exports.py
+++ b/trio/_tests/test_exports.py
@@ -9,6 +9,7 @@
 import sys
 from pathlib import Path
 from types import ModuleType
+from typing import Protocol
 
 import attrs
 import pytest
@@ -23,6 +24,12 @@
 mypy_cache_updated = False
 
 
+try:  # If installed, check both versions of this class.
+    from typing_extensions import Protocol as Protocol_ext
+except ImportError:  # pragma: no cover
+    Protocol_ext = Protocol
+
+
 def _ensure_mypy_cache_updated():
     # This pollutes the `empty` dir. Should this be changed?
     from mypy.api import run
@@ -311,12 +318,15 @@ def lookup_symbol(symbol):
             "__annotations__",
             "__attrs_attrs__",
             "__attrs_own_setattr__",
+            "__callable_proto_members_only__",
             "__class_getitem__",
+            "__final__",
             "__getstate__",
             "__match_args__",
             "__order__",
             "__orig_bases__",
             "__parameters__",
+            "__protocol_attrs__",
             "__setstate__",
             "__slots__",
             "__weakref__",
@@ -484,6 +494,9 @@ def test_classes_are_final():
             # point of ABCs
             if inspect.isabstract(class_):
                 continue
+            # Same with protocols, but only direct children.
+            if Protocol in class_.__bases__ or Protocol_ext in class_.__bases__:
+                continue
             # Exceptions are allowed to be subclassed, because exception
             # subclassing isn't used to inherit behavior.
             if issubclass(class_, BaseException):
diff --git a/trio/_tests/verify_types.json b/trio/_tests/verify_types.json
index ec345facf3..16acc7f39b 100644
--- a/trio/_tests/verify_types.json
+++ b/trio/_tests/verify_types.json
@@ -7,11 +7,11 @@
     "warningCount": 0
   },
   "typeCompleteness": {
-    "completenessScore": 0.9250398724082934,
+    "completenessScore": 0.9490445859872612,
     "exportedSymbolCounts": {
       "withAmbiguousType": 0,
-      "withKnownType": 580,
-      "withUnknownType": 47
+      "withKnownType": 596,
+      "withUnknownType": 32
     },
     "ignoreUnknownTypesFromImports": true,
     "missingClassDocStringCount": 1,
@@ -46,17 +46,11 @@
     ],
     "otherSymbolCounts": {
       "withAmbiguousType": 3,
-      "withKnownType": 602,
-      "withUnknownType": 61
+      "withKnownType": 627,
+      "withUnknownType": 50
     },
     "packageName": "trio",
     "symbols": [
-      "trio._core._run.Nursery.start",
-      "trio._core._run.Nursery.start_soon",
-      "trio._core._run.TaskStatus.__repr__",
-      "trio._core._run.TaskStatus.started",
-      "trio._dtls.DTLSChannel.__init__",
-      "trio._dtls.DTLSEndpoint.serve",
       "trio._highlevel_socket.SocketStream.getsockopt",
       "trio._highlevel_socket.SocketStream.send_all",
       "trio._highlevel_socket.SocketStream.setsockopt",
@@ -86,32 +80,22 @@
       "trio._subprocess.Process.send_signal",
       "trio._subprocess.Process.terminate",
       "trio._subprocess.Process.wait",
-      "trio.current_time",
       "trio.from_thread.run",
       "trio.from_thread.run_sync",
       "trio.lowlevel.cancel_shielded_checkpoint",
-      "trio.lowlevel.current_clock",
-      "trio.lowlevel.current_root_task",
-      "trio.lowlevel.current_statistics",
-      "trio.lowlevel.current_trio_token",
       "trio.lowlevel.notify_closing",
       "trio.lowlevel.open_process",
       "trio.lowlevel.permanently_detach_coroutine_object",
       "trio.lowlevel.reattach_detached_coroutine_object",
-      "trio.lowlevel.reschedule",
-      "trio.lowlevel.spawn_system_task",
-      "trio.lowlevel.start_guest_run",
       "trio.lowlevel.temporarily_detach_coroutine_object",
       "trio.lowlevel.wait_readable",
       "trio.lowlevel.wait_writable",
       "trio.open_ssl_over_tcp_listeners",
       "trio.open_ssl_over_tcp_stream",
       "trio.open_unix_socket",
-      "trio.run",
       "trio.run_process",
       "trio.serve_listeners",
       "trio.serve_ssl_over_tcp",
-      "trio.serve_tcp",
       "trio.testing._memory_streams.MemoryReceiveStream.__init__",
       "trio.testing._memory_streams.MemoryReceiveStream.aclose",
       "trio.testing._memory_streams.MemoryReceiveStream.close",
@@ -142,7 +126,6 @@
       "trio.testing.memory_stream_pump",
       "trio.testing.open_stream_to_socket_listener",
       "trio.testing.trio_test",
-      "trio.testing.wait_all_tasks_blocked",
       "trio.tests.TestsDeprecationWrapper",
       "trio.to_thread.current_default_thread_limiter"
     ]
diff --git a/trio/_tools/gen_exports.py b/trio/_tools/gen_exports.py
index f3ed2e26e7..9d78cd5bd7 100755
--- a/trio/_tools/gen_exports.py
+++ b/trio/_tools/gen_exports.py
@@ -158,6 +158,13 @@ def gen_public_wrappers_source(file: File) -> str:
         if is_cm:  # pragma: no cover
             func = func.replace("->Iterator", "->ContextManager")
 
+        # TODO: hacky workaround until we run mypy without `-m`, which breaks imports
+        # enough that it cannot figure out the type of _NO_SEND
+        if file.path.stem == "_run" and func.startswith(
+            "def reschedule"
+        ):  # pragma: no cover
+            func = func.replace("None:\n", "None:  # type: ignore[has-type]\n")
+
         # Create export function body
         template = TEMPLATE.format(
             " await " if isinstance(method, ast.AsyncFunctionDef) else " ",
@@ -247,7 +254,15 @@ def main() -> None:  # pragma: no cover
 
 
 IMPORTS_RUN = """\
-from ._run import _NO_SEND
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+from outcome import Outcome
+import contextvars
+
+from ._run import _NO_SEND, RunStatistics, Task
+from ._entry_queue import TrioToken
+from .._abc import Clock
 """
 IMPORTS_INSTRUMENT = """\
 from ._instrumentation import Instrument
diff --git a/trio/_util.py b/trio/_util.py
index ba56c18385..0329a12fa9 100644
--- a/trio/_util.py
+++ b/trio/_util.py
@@ -102,7 +102,7 @@ def is_main_thread() -> bool:
 # TODO: Use TypeVarTuple here.
 def coroutine_or_error(
     async_fn: t.Callable[..., t.Awaitable[RetT]], *args: t.Any
-) -> t.Awaitable[RetT]:
+) -> collections.abc.Coroutine[object, t.NoReturn, RetT]:
     def _return_value_looks_like_wrong_library(value: object) -> bool:
         # Returned by legacy @asyncio.coroutine functions, which includes
         # a surprising proportion of asyncio builtins.
diff --git a/trio/lowlevel.py b/trio/lowlevel.py
index c66e22b60e..25e64975e2 100644
--- a/trio/lowlevel.py
+++ b/trio/lowlevel.py
@@ -13,9 +13,9 @@
     ParkingLot as ParkingLot,
     ParkingLotStatistics as ParkingLotStatistics,
     RaiseCancelT as RaiseCancelT,
+    RunStatistics as RunStatistics,
     RunVar as RunVar,
     Task as Task,
-    TaskStatus as TaskStatus,
     TrioToken as TrioToken,
     UnboundedQueue as UnboundedQueue,
     UnboundedQueueStatistics as UnboundedQueueStatistics,
