diff --git a/docs/changelog.rst b/docs/changelog.rst
index fde66bd..8cf036e 100644
--- a/docs/changelog.rst
+++ b/docs/changelog.rst
@@ -4,6 +4,10 @@ Changelog
 
 `CalVer, YY.month.patch <https://calver.org/>`_
 
+26.8.1
+======
+- Add :ref:`ASYNC128 <async128>` task-status-never-started, warning about startable functions (i.e. with a ``task_status`` parameter) that never call ``task_status.started()``. `(issue #471) <https://github.com/python-trio/flake8-async/issues/471>`_
+
 26.7.1
 ======
 - Add :ref:`ASYNC401 <async401>` pytest-raises-exception-group, recommending ``pytest.RaisesGroup`` over ``pytest.raises(ExceptionGroup)``. `(issue #430) <https://github.com/python-trio/flake8-async/issues/430>`_
diff --git a/docs/rules.rst b/docs/rules.rst
index f7c3ee1..2082bcc 100644
--- a/docs/rules.rst
+++ b/docs/rules.rst
@@ -141,6 +141,18 @@ _`ASYNC127`: unmaintained-httpx
     migration is usually just a matter of replacing ``httpx`` with ``httpx2`` in
     imports.  This rule triggers on any import of ``httpx``.
 
+_`ASYNC128`: task-status-never-started
+    A startable function - one taking a ``task_status`` keyword parameter, or with a
+    parameter annotated as ``TaskStatus`` - never calls ``task_status.started()``.
+    :meth:`trio.Nursery.start`/:meth:`anyio.abc.TaskGroup.start` wait for the callee
+    to call ``task_status.started()``: if it returns without doing so they raise
+    ``RuntimeError``, and if it never returns they block until cancelled.
+    This check is intentionally strict: passing ``task_status`` on to a helper
+    function or assigning it to another variable does not count, only a direct call
+    in the function body, or in a nested function where the parameter isn't shadowed.
+    Functions with stub bodies (only ``pass``, ``...``, string constants, and/or
+    ``raise``) are ignored, e.g. overloads, protocols, and abstract methods.
+
 Blocking sync calls in async functions
 ======================================
 
diff --git a/docs/usage.rst b/docs/usage.rst
index 10c11b1..00e092f 100644
--- a/docs/usage.rst
+++ b/docs/usage.rst
@@ -33,7 +33,7 @@ adding the following to your ``.pre-commit-config.yaml``:
    minimum_pre_commit_version: '2.9.0'
    repos:
    - repo: https://github.com/python-trio/flake8-async
-     rev: 26.7.1
+     rev: 26.8.1
      hooks:
        - id: flake8-async
          # args: ["--enable=ASYNC100,ASYNC112", "--disable=", "--autofix=ASYNC"]
diff --git a/flake8_async/__init__.py b/flake8_async/__init__.py
index f3e6a47..ab66ef4 100644
--- a/flake8_async/__init__.py
+++ b/flake8_async/__init__.py
@@ -38,7 +38,7 @@
 
 
 # CalVer: YY.month.patch, e.g. first release of July 2022 == "22.7.1"
-__version__ = "26.7.1"
+__version__ = "26.8.1"
 
 
 # taken from https://github.com/Zac-HD/shed
diff --git a/flake8_async/visitors/visitors.py b/flake8_async/visitors/visitors.py
index 5a66d89..281d613 100644
--- a/flake8_async/visitors/visitors.py
+++ b/flake8_async/visitors/visitors.py
@@ -16,7 +16,7 @@
 )
 
 if TYPE_CHECKING:
-    from collections.abc import Mapping
+    from collections.abc import Iterable, Mapping
 
     import libcst as cst
 
@@ -617,6 +617,59 @@ def visit_ImportFrom(self, node: ast.ImportFrom):
             self.error(node)
 
 
+@error_class
+class Visitor128(Flake8AsyncVisitor):
+    error_codes: Mapping[str, str] = {
+        "ASYNC128": (
+            "Startable function {} never calls `{}.started()`, so `.start()`"
+            " calls on it will fail, or block forever."
+        ),
+    }
+
+    # Look for a `<name>.started()` call anywhere in the function body, including
+    # in nested functions closing over the parameter. Nested functions that rebind
+    # the name are startable functions of their own, checked separately.
+    def _calls_started(self, node: ast.AST, name: str) -> bool:
+        if isinstance(node, ast.Call) and ast.unparse(node.func) == f"{name}.started":
+            return True
+        children: Iterable[ast.AST] = ast.iter_child_nodes(node)
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda)):
+            a = node.args
+            if any(
+                p is not None and p.arg == name
+                for p in (*a.posonlyargs, *a.args, *a.kwonlyargs, a.vararg, a.kwarg)
+            ):
+                return False
+            # only look in the body - decorators, defaults and annotations are
+            # evaluated in the enclosing scope, but a call in them is nonsensical
+            children = [node.body] if isinstance(node, ast.Lambda) else node.body
+        return any(self._calls_started(child, name) for child in children)
+
+    def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef):
+        args = node.args
+        for arg in (*args.posonlyargs, *args.args, *args.kwonlyargs):
+            # startable: a `task_status` parameter - unless positional-only, when it
+            # can't be passed by keyword - or a `TaskStatus`-annotated parameter
+            ann = arg.annotation
+            if isinstance(ann, ast.Subscript):  # strip generics: `TaskStatus[int]`
+                ann = ann.value
+            if not (
+                (
+                    ann is not None
+                    and ast.unparse(ann).rsplit(".", 1)[-1] == "TaskStatus"
+                )
+                or (arg.arg == "task_status" and arg not in args.posonlyargs)
+            ):
+                continue
+            if not all(  # stub bodies are fine: overloads, protocols, abstractmethods
+                isinstance(stmt, (ast.Pass, ast.Raise))
+                or (isinstance(stmt, ast.Expr) and isinstance(stmt.value, ast.Constant))
+                for stmt in node.body
+            ) and not any(self._calls_started(stmt, arg.arg) for stmt in node.body):
+                self.error(node, node.name, arg.arg)
+            return
+
+
 @error_class_cst
 class Visitor300(Flake8AsyncVisitor_cst):
     error_codes: Mapping[str, str] = {
diff --git a/tests/eval_files/async128.py b/tests/eval_files/async128.py
new file mode 100644
index 0000000..e4f5498
--- /dev/null
+++ b/tests/eval_files/async128.py
@@ -0,0 +1,171 @@
+"""Test file for ASYNC128 task-status-never-started."""
+
+# ASYNC128 does not care about the imported library, so will raise errors regardless
+# of trio/anyio/asyncio
+
+from typing import Any
+
+import trio
+from trio import TaskStatus
+
+
+async def never_started(  # error: 0, "never_started", "task_status"
+    task_status=trio.TASK_STATUS_IGNORED,
+):
+    await trio.sleep(1)
+
+
+async def started(task_status=trio.TASK_STATUS_IGNORED):
+    task_status.started()
+
+
+async def started_with_value(task_status=trio.TASK_STATUS_IGNORED):
+    task_status.started(7)
+
+
+async def no_task_status():
+    await trio.sleep(1)
+
+
+# a conditional call counts - no attempt is made to check all code paths
+async def conditional_start(condition: bool, *, task_status):
+    if condition:
+        task_status.started()
+
+
+# ... even if the call can never actually execute
+async def unreachable_start(task_status):
+    for _ in range(0):
+        task_status.started()
+
+
+# annotated parameters trigger regardless of their name
+async def annotated(status: TaskStatus[int]):  # error: 0, "annotated", "status"
+    await trio.sleep(1)
+
+
+async def annotated_bare(status: TaskStatus):  # error: 0, "annotated_bare", "status"
+    await trio.sleep(1)
+
+
+async def annotated_qualified(  # error: 0, "annotated_qualified", "status"
+    status: trio.TaskStatus[int],
+):
+    await trio.sleep(1)
+
+
+async def annotated_ok(status: TaskStatus[int]):
+    status.started(5)
+
+
+# a `task_status` parameter that's positional-only can't be startable, but an
+# explicit annotation still counts
+async def posonly_ignored(task_status, /):
+    await trio.sleep(1)
+
+
+async def posonly_annotated(  # error: 0, "posonly_annotated", "status"
+    status: TaskStatus[int], /
+):
+    await trio.sleep(1)
+
+
+async def starargs_ignored(*task_status, **kwargs):
+    await trio.sleep(1)
+
+
+# passing `task_status` to a helper does not count, even if the helper calls
+# `started()` for you. The check is intentionally strict, silence it with `noqa`
+# if you're intentionally proxying it.
+async def helper(fn: Any, task_status):  # error: 0, "helper", "task_status"
+    await fn(task_status=task_status)
+
+
+# aliasing does not count either
+async def aliased(task_status):  # error: 0, "aliased", "task_status"
+    ts = task_status
+    ts.started()
+
+
+# accessing `.started` without calling it does not count
+async def not_called(task_status):  # error: 0, "not_called", "task_status"
+    task_status.started
+
+
+# the call must be on the parameter itself, not e.g. an attribute by the same name
+class AttributeStatus:
+    task_status: TaskStatus[None]
+
+    async def relay(self, task_status):  # error: 4, "relay", "task_status"
+        self.task_status.started()
+
+
+# calls in nested functions closing over the parameter do count
+async def closure(task_status=trio.TASK_STATUS_IGNORED):
+    def inner():
+        task_status.started()
+
+    inner()
+
+
+async def lambda_closure(task_status=trio.TASK_STATUS_IGNORED):
+    fn = lambda: task_status.started()
+    fn()
+
+
+# ... but not if the nested function rebinds the name; it is instead
+# checked on its own
+async def shadowed(task_status):  # error: 0, "shadowed", "task_status"
+    async def inner(task_status=trio.TASK_STATUS_IGNORED):
+        task_status.started()
+
+    await inner()
+
+
+async def shadowed_by_lambda(  # error: 0, "shadowed_by_lambda", "task_status"
+    task_status,
+):
+    fn = lambda task_status: task_status.started()
+    fn(None)
+
+
+async def shadowed_by_vararg(  # error: 0, "shadowed_by_vararg", "task_status"
+    task_status,
+):
+    def inner(*task_status: Any):
+        task_status[0].started()
+
+    inner(None)
+
+
+async def nested_never_started():
+    async def inner(  # error: 4, "inner", "task_status"
+        task_status=trio.TASK_STATUS_IGNORED,
+    ):
+        await trio.sleep(1)
+
+    await inner()
+
+
+# stub bodies don't error, e.g. overloads, protocols, and abstract methods
+class StartableProtocol:
+    async def ellipsis_body(self, *, task_status: TaskStatus[None]): ...
+
+    async def pass_body(self, *, task_status: TaskStatus[None]):
+        pass
+
+    async def docstring_body(self, *, task_status: TaskStatus[None]):
+        """It has a docstring."""
+
+    async def raise_body(self, *, task_status: TaskStatus[None]):
+        raise NotImplementedError
+
+    async def method_never_started(  # error: 4, "method_never_started", "task_status"
+        self, task_status=trio.TASK_STATUS_IGNORED
+    ):
+        await trio.sleep(1)
+
+
+# sync functions are not checked - they cannot be passed to `.start()`
+def sync_fn(task_status):
+    return None