diff --git a/docs/conf.py b/docs/conf.py
index 0325064f3..7feccb076 100644
--- a/docs/conf.py
+++ b/docs/conf.py
@@ -111,6 +111,10 @@
 nitpick_ignore_regex = [
     ("py:class", r"fastcs.*.DType_T"),
     ("py:class", r"fastcs.*.Numeric_T"),
+    ("py:class", r"fastcs.*.Introspection_T"),
+    ("py:class", r"fastcs.*.Connection_T"),
+    ("py:obj", r"fastcs.*.Introspection_T"),
+    ("py:obj", r"fastcs.*.Connection_T"),
     ("py:obj", r"fastcs.*.DType_T"),
     (r"py:.*", r"fastcs\.demo.*"),
     (r"py:.*", r"tickit.*"),
diff --git a/docs/explanations/connections.md b/docs/explanations/connections.md
new file mode 100644
index 000000000..83ed9e36f
--- /dev/null
+++ b/docs/explanations/connections.md
@@ -0,0 +1,182 @@
+# Connections
+
+A `Connection` is a link to hardware, and it owns its own health state. Controllers
+hold a connection; several controllers may hold the same one.
+
+Connections, not controllers, are the unit of failure and recovery. A tree of five
+sub controllers behind one socket has one health state, one reconnect task and one
+retry budget between them - not five of each, four of which can do nothing about the
+link that is actually down.
+
+## Writing one
+
+Subclass `Connection`, open the link in `connect` and close it in `close`:
+
+```python
+from dataclasses import dataclass
+
+from fastcs.connections import Connection
+
+
+@dataclass
+class DetectorInfo:
+    """Returned by connect(). Compared against the startup value on every
+    reconnect, so it must compare by value - hence the dataclass."""
+
+    api_version: str
+    parameters: tuple[str, ...]
+
+
+class DetectorConnection(Connection[DetectorInfo]):
+    # Class defaults sit between the framework defaults and any constructor argument.
+    reconnect_period = 5.0
+    max_attempts = 60
+
+    def __init__(self, settings: IPConnectionSettings, **kwargs) -> None:
+        super().__init__(**kwargs)
+        self._settings = settings
+        self._client: AsyncClient | None = None
+
+    async def connect(self) -> DetectorInfo:
+        base = f"http://{self._settings.ip}:{self._settings.port}"
+        self._client = AsyncClient(base_url=base)
+        return DetectorInfo(
+            api_version=await self.get("detector/api/version"),
+            parameters=tuple(await self.get("detector/api/1.8.0/config/keys")),
+        )
+
+    async def close(self) -> None:
+        if self._client is not None:
+            await self._client.aclose()
+            self._client = None
+
+    async def get(self, path: str):
+        try:
+            response = await self._client.get(path)
+        except (ConnectError, ReadTimeout):
+            # The transport is gone. Everything holding this connection is now down.
+            self.set_disconnected()
+            raise
+        # A 400 from the detector is a device complaint, not a dead link - it
+        # propagates to the caller without touching connection state.
+        response.raise_for_status()
+        return response.json()["value"]
+```
+
+`connect` means "make the link usable", not merely "open the socket": a device that
+needs a mode set before it can be introspected has that write here.
+
+**The important part is the `except` clause.** The connection is the only place that
+can tell "the socket died" from "the device rejected that parameter", and only the
+first is a connection failure. Nothing above a connection has to catch anything, and
+no exception type is a contract between layers.
+
+**The framework sets the state; authors do the work and raise.** No driver touches a
+connected flag: `connect` opens the link or raises, and the framework decides what
+that means. The one thing a driver calls is `set_disconnected`, from its own IO.
+
+## Holding one
+
+A controller claims a connection by name from the `Connections` registry, which is
+forwarded down the tree. Passing the registry rather than a bare connection means a
+controller's constructor signature does not change when something three tiers below
+it needs a new connection:
+
+```python
+class DetectorController(Controller):
+    # Narrows the base class's connection so this controller's own code can call
+    # the methods of the connection it actually holds.
+    connection: DetectorConnection
+
+    def __init__(self, connections: Connections) -> None:
+        # Claimed by name, type asserted. Raises at construction - before anything
+        # opens - if the name is missing or the type is wrong.
+        self.connection = connections.get("detector", DetectorConnection)
+        super().__init__()
+
+    async def build(self, info: DetectorInfo) -> None:
+        for parameter in info.parameters:
+            ...  # one attribute per reported key
+```
+
+A controller holds at most one connection - two devices means two controllers. A
+controller with no connection at all (a soft controller that only groups others, or a
+`ControllerVector`) is never gated and never reconnected.
+
+**No controller ever reads another controller's state.** A sub controller that shares
+its parent's connection is not consulting its parent - it holds the same object.
+Failure, gating and recovery all resolve through that shared object, never through
+the tree.
+
+## Startup
+
+The `ControllerRunner` owns the order:
+
+1. Open every connection, in declaration order, keeping what `connect` returned.
+2. Walk the tree calling `build`, repeating over anything newly added until a pass
+   adds nothing.
+3. Call `setup` across the whole built tree.
+4. Warn about anything suspicious, run the initial reads, and start the tasks.
+
+A failure anywhere in startup aborts. A partly built tree means an application with a
+silently incomplete set of parameters, which is worse than no application at all,
+because clients connect successfully and never find what they are looking for. The
+orchestrator owns the retry.
+
+## Failure and recovery
+
+Failure is detected in exactly one place: the connection's own IO. `set_disconnected`
+wakes that connection's reconnect task and gates every scan that uses it.
+
+There is one reconnect task per connection, idle until that connection actually goes
+down - a healthy connection costs nothing, and each connection recovers at its own
+pace. A detector that wants to retry every five seconds does not have to compromise
+with a writer that wants one.
+
+Each attempt closes the link, reopens it, and compares what `connect` returned
+against the startup value. `max_attempts` consecutive failures is terminal until the
+process restarts; a clean connection restores the budget.
+
+### Dependencies
+
+A connection layered over another declares it, rather than having it derived from
+where controllers sit in the tree:
+
+```python
+odin = OdinConnection(settings, depends_on=detector)
+```
+
+While the dependency is down, the dependent waits instead of attempting - and because
+no attempt means no increment, its retry budget freezes rather than being burnt
+against a dead dependency. If the dependency gives up entirely, the dependent is
+released rather than left hanging: it logs that it is stalled and waits for a
+restart. Cycles are caught at startup.
+
+### Introspection is checked, not re-applied
+
+`build` cannot run again, so a device that comes back describing itself differently
+cannot be accommodated. Rather than carrying on against a structure that no longer
+matches the hardware, the runner records the mismatch and `FastCS.serve` raises it -
+an explicit, observable failure rather than a `sys.exit` an embedder cannot survive.
+
+Because the comparison is `!=`, an introspection result has to compare to a single
+bool. A dataclass of plain fields does; an array of values does not, and the runner
+says so rather than letting an ambiguous truth value escape from a background task.
+
+## Warnings
+
+- A connection declared but never claimed is warned about at startup: it would
+  otherwise be opened and reconnected forever while doing nothing.
+- A connection with no polled attribute or scan method among any of its controllers
+  is warned about, phrased as fact rather than fault - all-on-demand is a legitimate
+  design, it just means nothing will detect the link failing until the next write.
+
+There is no separate health-check hook. A connection with any polling is proved alive
+by that polling; a device that genuinely needs a heartbeat gets a `@scan` on one of
+its controllers, which is ordinary driver code.
+
+## Shutdown
+
+Closing is a runner operation, not an author hook: every connection is closed in
+reverse declaration order, so anything layered over another is closed before what it
+rides on. `setup` is not undone - devices keep their last configured state.
diff --git a/docs/explanations/controllers.md b/docs/explanations/controllers.md
index 30a09da0a..f4e4a287a 100644
--- a/docs/explanations/controllers.md
+++ b/docs/explanations/controllers.md
@@ -7,32 +7,49 @@ FastCS provides three controller classes: `Controller`, `ControllerVector`, and
 
 `Controller` is the primary building block for FastCS drivers. It can serve two roles:
 
-**Root controller:** passed directly to the `FastCS` launcher. In this role, FastCS
-will call its lifecycle hooks and run the scan tasks it creates on the event loop.
+**Root controller:** passed directly to the `FastCS` launcher.
 
 **Sub controller:** attached to a parent controller via `add_sub_controller()` or by
-assigning it as an attribute. In this role, the sub controller's lifecycle hooks
-(`connect`, `reconnect`, `initialise`, `disconnect`) are not called automatically by
-FastCS. The parent controller is responsible for calling them as part of its own
-lifecycle, if required.
+assigning it as an attribute.
+
+The `ControllerRunner` owns the order of the startup sequence and calls the
+lifecycle hooks of **every** controller in the tree, root and sub alike. A parent
+never drives a child's lifecycle to compensate for sequencing.
 
 ### Lifecycle hooks
 
 | Method | Purpose |
 |---|---|
-| `initialise` | Dynamically add attributes on startup, before the API is built |
-| `connect` | Open connection to device |
-| `reconnect` | Re-open connection after scan error |
-| `disconnect` | Release device resources before shutdown |
+| `__init__` | Everything knowable without the device: settings, static attributes |
+| `build` | Structure that depends on the device - attributes and sub controllers |
+| `setup` | Hardware writes and checks, once the whole tree is built |
+
+The same question, three ways:
+
+| What do I need to answer this? | Where it goes |
+|---|---|
+| Nothing - settings and the class | `__init__` |
+| The device | `build` |
+| My children, connected | `setup` |
+
+There is no `connect`, `reconnect` or `disconnect` hook. Opening the link, reopening
+it after a failure and closing it at shutdown belong to the `Connection` and the
+runner - see [connections](./connections.md).
+
+`build` optionally receives whatever its connection's `connect` returned: write
+`build(self)` for nothing, or `build(self, info)` to be handed the connection's
+introspection result.
 
 ### Scan task behaviour
 
-When used as the root controller, FastCS collects all `@scan` methods and readable
-attributes whose `getter` is wrapped in `Polled`, across the whole controller
-hierarchy, to be run as background tasks by FastCS. Scan tasks are gated on the
-`_connected` flag: if a scan
-raises an exception, `_connected` is set to `False` and tasks pause until `reconnect`
-sets it back to `True`.
+FastCS collects all `@scan` methods and readable attributes whose `getter` is wrapped
+in `Polled`, across the whole controller hierarchy, to be run as background tasks.
+Scan tasks are gated on the controller's **connection**: while that connection is
+down they wait for it to come back rather than polling a link that cannot answer. A
+controller with no connection is never gated.
+
+A scan that raises is logged and retried. It does not mark the connection down -
+only the connection's own IO can tell a dead transport from a device complaint.
 
 ```python
 from fastcs.controllers import Controller
@@ -41,56 +58,54 @@ from fastcs.methods import scan
 
 
 class TemperatureController(Controller):
+    connection: DeviceConnection
+
     temperature = AttrR(float, units="degC")
     setpoint = AttrRW(float, units="degC")
 
-    async def connect(self):
-        self._client = await DeviceClient.connect(self._host, self._port)
-        self._connected = True
-
-    async def reconnect(self):
-        try:
-            self._client = await DeviceClient.connect(self._host, self._port)
-            self._connected = True
-        except Exception:
-            logger.error("Failed to reconnect")
-
-    async def disconnect(self):
-        await self._client.close()
+    def __init__(self, connections: Connections):
+        self.connection = connections.get("device", DeviceConnection)
+        super().__init__()
 
     @scan(period=1.0)
     async def update_temperature(self):
-        value = await self._client.get_temperature()
+        # Gated on the connection: while it is down this does not run at all.
+        value = await self.connection.get_temperature()
         await self.temperature.update(value)
 ```
 
 ### Using Controller as a sub controller
 
 When a `Controller` is nested inside another, it organises the driver into logical
-sections and its attributes are exposed under a prefixed path. If the sub
-controller also has connection logic, the parent must invoke it explicitly:
+sections and its attributes are exposed under a prefixed path. A sub controller that
+talks to the same device holds the *same* connection object as its parent rather
+than consulting it, so the two share one health state and one reconnect task:
 
 ```python
 class ChannelController(Controller):
+    connection: DeviceConnection
+
     value = AttrR(float)
 
-    async def connect(self):
-        ...
-        self._connected = True
+    def __init__(self, connection: DeviceConnection):
+        self.connection = connection
+        super().__init__()
 
 
 class RootController(Controller):
+    connection: DeviceConnection
+
     channel: ChannelController
 
-    def __init__(self):
+    def __init__(self, connections: Connections):
+        self.connection = connections.get("device", DeviceConnection)
         super().__init__()
-        self.channel = ChannelController()
-
-    async def connect(self):
-        await self.channel.connect()
-        self._connected = True
+        self.channel = ChannelController(self.connection)
 ```
 
+A sub controller that talks to a *different* device claims its own connection by
+name from the registry instead. Nothing is inferred from tree position.
+
 ## ControllerVector
 
 `ControllerVector` is a convenience wrapper for a set of controllers of the same type,
@@ -120,12 +135,6 @@ class RootController(Controller):
             {i: ChannelController() for i in range(num_channels)}
         )
 
-    async def connect(self):
-        for channel in self.channels.values():
-            await channel.connect()
-
-        self._connected = True
-
     async def update_all(self):
         for index, channel in self.channels.items():
             value = await self._client.get_channel(index)
@@ -145,7 +154,7 @@ Use `ControllerVector` when:
 
 - The device has a set of identical channels, axes, or modules identified by number
 - You need to iterate over sub controllers and perform the same action on each
-- The number of instances may vary (e.g. determined at runtime during `initialise`)
+- The number of instances may vary (e.g. determined at runtime during `build`)
 
 Use a plain `Controller` with named sub controllers when the sub controllers are
 distinct components with different types or roles.
diff --git a/docs/explanations/decisions/0016-setpoint-cache-timestamps-and-controller-runner.md b/docs/explanations/decisions/0016-setpoint-cache-timestamps-and-controller-runner.md
index 4f67b0528..95e694247 100644
--- a/docs/explanations/decisions/0016-setpoint-cache-timestamps-and-controller-runner.md
+++ b/docs/explanations/decisions/0016-setpoint-cache-timestamps-and-controller-runner.md
@@ -118,3 +118,50 @@ surface referenced by decision 13 of #388.
    responsibility**.
 4. **Who owns reconnect?** The runner owns the whole lifecycle, including
    reconnect.
+
+## Amendment: connections own reconnect (#422)
+
+Question 4 above says the runner owns reconnect, and stops there. It left the
+*subject* of reconnect implicit, and the first implementation took it to be the
+controller — which cannot work once controllers share a link, because
+reconnecting a controller that does not own its connection is a no-op.
+
+The subject is the **connection**. Connection state moves off `Controller` onto a
+first-class `Connection` object: controllers hold one, several may hold the same
+one, and it owns its own health, reconnect task and retry budget. See
+[connections](../connections.md) for the shape, and the design attached to
+[issue #422](https://github.com/DiamondLightSource/fastcs/issues/422).
+
+What this amends:
+
+- **`Controller.connect`/`reconnect`/`disconnect`/`_connected` are removed.** A
+  driver never touches a connected flag; `Connection.connect()` opens the link or
+  raises, and the framework decides what that means.
+- **`initialise`/`post_initialise` become `build`/`setup`**, splitting "structure
+  that depends on the device" from "hardware writes once the tree is built".
+  `build` optionally receives whatever the connection's `connect()` returned.
+- **Introspection is checked on every reconnect.** `build` runs once, so a device
+  that comes back describing itself differently cannot be accommodated.
+- **The runner owns the startup order** — connections, then `build` to a fixpoint,
+  then `setup`, then tasks — and shutdown, closing connections in reverse.
+
+Points the review of #420 asked to settle, and how they land:
+
+- **`connection._connected = True` from outside.** There is a framework-only
+  `_set_connected()` next to `set_disconnected()`, so the flag has one owner.
+- **What "fatal" means for an introspection mismatch.** Not `sys.exit`: an
+  embedded FastCS must survive it. The runner sets `fatal_error` (an
+  `asyncio.Event`) and records `fatal_reason`; `FastCS.serve` raises it, and an
+  embedder observes it instead.
+- **How build info is compared.** With `!=`, which means an introspection result
+  must compare to a single bool. An ambiguous comparison (a dict of numpy arrays)
+  raises a message saying so rather than escaping a background task.
+- **`check()` skipped while IO is succeeding.** Dropped. There is no separate
+  health-check hook: a connection with any polling is proved alive by that
+  polling, and a device that needs a heartbeat gets a `@scan`, which is ordinary
+  driver code. A connection nothing polls is warned about at startup.
+- **`depends_on` cycles.** Declared rather than derived, and detected at startup.
+- **`max_attempts` exhausting to a terminal state.** Kept as the design specifies
+  (default 10), and *not* propagated to dependents — the parent's give-up message
+  names what it blocks. Whether the default should instead be retry-forever is
+  left open; it is one constant.
diff --git a/docs/explanations/stable-interface.md b/docs/explanations/stable-interface.md
index 52f9c4b21..1c551f4bb 100644
--- a/docs/explanations/stable-interface.md
+++ b/docs/explanations/stable-interface.md
@@ -14,30 +14,40 @@ transports, no interactive shell. `FastCS` is a caller of it.
 ```python
 from fastcs.controllers import ControllerRunner
 
-runner = ControllerRunner(controller)
-apis = await runner.setup()   # initialise, and build the ControllerAPIs
-await runner.start()          # connect, run initial tasks, start scanning
+runner = ControllerRunner(controller, connections=connections)
+apis = await runner.build()   # open connections, build the tree, build the APIs
+await runner.start()          # setup, run initial tasks, start scanning
 ...
-await runner.stop()           # stop the tasks, disconnect
+await runner.stop()           # stop the tasks, close the connections
 ```
 
-- **`setup()`** runs `initialise()` and `post_initialise()` on each controller
-  and builds their `ControllerAPI`s. It exists as a separate step because
-  anything serving the controllers has to register its callbacks *before* the
-  first values are read, or it misses them.
-- **`start()`** connects the controllers, runs the initial (`ONCE`) tasks, and
-  starts the periodic ones. It runs `setup()` first if you have not, so an
-  embedder that does not need the APIs in between can just call `start()`.
-- **`stop()`** cancels the tasks and disconnects.
+- **`build()`** opens every declared `Connection`, walks the tree calling
+  `build()` on each controller, and builds their `ControllerAPI`s. It exists as
+  a separate step because anything serving the controllers has to register its
+  callbacks *before* the first values are read, or it misses them.
+- **`start()`** runs `setup()` across the tree, the initial (`ONCE`) tasks, and
+  then starts the periodic ones and one reconnect task per connection. It runs
+  `build()` first if you have not, so an embedder that does not need the APIs in
+  between can just call `start()`.
+- **`stop()`** cancels the tasks and closes every connection, in reverse
+  declaration order.
 
 **Idempotency is the caller's responsibility.** Starting a running runner, or
 stopping a stopped one, is not defined — an embedder whose own connect may run
 more than once has to keep track itself.
 
-The runner also owns **reconnect**. A scan task whose callback raises marks its
-controller disconnected and pauses rather than dying; the runner notices and
-calls `Controller.reconnect()` until it comes back. This is deliberately not
-left to each controller, so every controller recovers the same way.
+The runner also owns **reconnect**, per connection rather than per controller:
+see [connections](./connections.md). A connection's own IO marks it down when its
+transport fails, and that connection's reconnect task brings it back at its own
+pace. This is deliberately not left to each controller, so every connection
+recovers the same way and controllers sharing one recover together.
+
+**A fatal runner condition is observable rather than fatal to the process.**
+`runner.fatal_error` is an `asyncio.Event` set when the runner cannot carry on —
+a device coming back from a reconnect describing itself differently, say — with
+`runner.fatal_reason` carrying why. Nothing calls `sys.exit`, so an embedded
+FastCS inside another process decides for itself what to do; `FastCS.serve`
+raises it.
 
 ## Reading the structure: `ControllerAPI`
 
diff --git a/docs/how-to/update-attributes-from-device.md b/docs/how-to/update-attributes-from-device.md
index b5b710965..56f6fd3c2 100644
--- a/docs/how-to/update-attributes-from-device.md
+++ b/docs/how-to/update-attributes-from-device.md
@@ -165,8 +165,11 @@ class MultiChannelController(Controller):
             await channel.voltage.update(float(voltage))
 ```
 
-The scan period (here `0.1` seconds) sets how often the batched query runs. Scans that
-raise an exception will pause and wait for `reconnect()` to be called before resuming.
+The scan period (here `0.1` seconds) sets how often the batched query runs. A scan that
+raises is logged and tried again on the next period. If the failure was the connection
+itself going down, the scan waits for the connection to come back up rather than
+querying a dead link - the runner reopens it, and there is nothing for a driver to
+call. See [](../explanations/connections.md).
 
 ### Scan as a cache for getters
 
@@ -236,13 +239,13 @@ class SubscriptionController(Controller):
         super().__init__()
         self._client = subscription_client
 
-    async def connect(self):
+    async def setup(self):
         # Register an async callback that forwards updates into the attribute.
+        # `setup` runs once the whole tree is built and every connection is open.
         async def on_temperature_change(value: float) -> None:
             await self.temperature.update(value)
 
         await self._client.subscribe("temperature", on_temperature_change)
-        await super().connect()
 ```
 
 If the library only supports synchronous callbacks, schedule the coroutine onto the
diff --git a/docs/snippets/dynamic.py b/docs/snippets/dynamic.py
index 4e7c00e72..e07725099 100644
--- a/docs/snippets/dynamic.py
+++ b/docs/snippets/dynamic.py
@@ -80,17 +80,23 @@ async def setter(value, command=command, dtype=datatype):
 
 
 class TemperatureRampController(Controller):
+    connection: IPConnection
+
     def __init__(
         self,
         index: int,
         parameters: dict[str, TemperatureControllerParameter],
         protocol: TemperatureProtocol,
+        connection: IPConnection,
     ):
         self._parameters = parameters
         self._protocol = protocol
+        # The same connection the parent holds, so this controller's polled
+        # attributes pause with it while it is down.
+        self.connection = connection
         super().__init__(f"Ramp{index}")
 
-    async def initialise(self):
+    async def build(self):
         for name, attribute in create_attributes(
             self._parameters, self._protocol
         ).items():
@@ -98,20 +104,20 @@ async def initialise(self):
 
 
 class TemperatureController(Controller):
+    connection: IPConnection
+
     def __init__(self, settings: IPConnectionSettings):
-        self._ip_settings = settings
-        self._connection = IPConnection()
-        self._protocol = TemperatureProtocol(self._connection)
+        # Opening it, and reopening it after a failure, is the runner's job.
+        self.connection = IPConnection(settings)
+        self._protocol = TemperatureProtocol(self.connection)
 
         super().__init__()
 
-    async def connect(self):
-        await self._connection.connect(self._ip_settings)
-
-    async def initialise(self):
-        await self.connect()
-
-        api = json.loads((await self._connection.send_query("API?\r\n")).strip("\r\n"))
+    async def build(self):
+        # Runs with the connection already open. The ramp controllers added here get
+        # their own `build` called by the runner on a later pass, so there is no
+        # need - and no way - to drive their lifecycle from this one.
+        api = json.loads((await self.connection.send_query("API?\r\n")).strip("\r\n"))
 
         ramps_api = api.pop("Ramps")
 
@@ -120,13 +126,10 @@ async def initialise(self):
 
         for idx, ramp_parameters in enumerate(ramps_api):
             ramp_controller = TemperatureRampController(
-                idx + 1, ramp_parameters, self._protocol
+                idx + 1, ramp_parameters, self._protocol, self.connection
             )
-            await ramp_controller.initialise()
             self.add_sub_controller(f"Ramp{idx + 1:02d}", ramp_controller)
 
-        await self._connection.close()
-
 
 epics_ca = EpicsCATransport()
 connection_settings = IPConnectionSettings("localhost", 25565)
diff --git a/docs/snippets/static06.py b/docs/snippets/static06.py
index f7bd33d15..7938fbf84 100644
--- a/docs/snippets/static06.py
+++ b/docs/snippets/static06.py
@@ -9,16 +9,14 @@
 
 
 class TemperatureController(Controller):
+    connection: IPConnection
+
     device_id = AttrR(str)
 
     def __init__(self, settings: IPConnectionSettings):
         super().__init__()
 
-        self._ip_settings = settings
-        self._connection = IPConnection()
-
-    async def connect(self):
-        await self._connection.connect(self._ip_settings)
+        self.connection = IPConnection(settings)
 
 
 gui_options = EpicsGUIOptions(output_dir=Path("."), title="Demo Temperature Controller")
diff --git a/docs/snippets/static07.py b/docs/snippets/static07.py
index 2aea3bb76..6752c295f 100644
--- a/docs/snippets/static07.py
+++ b/docs/snippets/static07.py
@@ -9,21 +9,19 @@
 
 
 class TemperatureController(Controller):
+    connection: IPConnection
+
     def __init__(self, settings: IPConnectionSettings):
-        self._ip_settings = settings
-        self._connection = IPConnection()
+        self.connection = IPConnection(settings)
 
         super().__init__()
 
         self.device_id = AttrR(str, getter=Polled(self._get_device_id, period=0.2))
 
     async def _get_device_id(self) -> str:
-        response = await self._connection.send_query("ID?\r\n")
+        response = await self.connection.send_query("ID?\r\n")
         return response.strip("\r\n")
 
-    async def connect(self):
-        await self._connection.connect(self._ip_settings)
-
 
 gui_options = EpicsGUIOptions(output_dir=Path("."), title="Demo Temperature Controller")
 epics_ca = EpicsCATransport(gui=gui_options)
diff --git a/docs/snippets/static08.py b/docs/snippets/static08.py
index 95363380b..9e6557cf6 100644
--- a/docs/snippets/static08.py
+++ b/docs/snippets/static08.py
@@ -27,10 +27,11 @@ async def send_query(self, param: str, dtype: type[ValueT]) -> ValueT:
 
 
 class TemperatureController(Controller):
+    connection: IPConnection
+
     def __init__(self, settings: IPConnectionSettings):
-        self._ip_settings = settings
-        self._connection = IPConnection()
-        self._protocol = TemperatureProtocol(self._connection)
+        self.connection = IPConnection(settings)
+        self._protocol = TemperatureProtocol(self.connection)
 
         super().__init__()
 
@@ -43,9 +44,6 @@ async def _get_device_id(self) -> str:
     async def _get_power(self) -> float:
         return await self._protocol.send_query("P", float)
 
-    async def connect(self):
-        await self._connection.connect(self._ip_settings)
-
 
 gui_options = EpicsGUIOptions(output_dir=Path("."), title="Demo Temperature Controller")
 epics_ca = EpicsCATransport(gui=gui_options)
diff --git a/docs/snippets/static09.py b/docs/snippets/static09.py
index 10c07b334..f9618598e 100644
--- a/docs/snippets/static09.py
+++ b/docs/snippets/static09.py
@@ -27,10 +27,11 @@ async def send_query(self, param: str, dtype: type[ValueT]) -> ValueT:
 
 
 class TemperatureController(Controller):
+    connection: IPConnection
+
     def __init__(self, settings: IPConnectionSettings):
-        self._ip_settings = settings
-        self._connection = IPConnection()
-        self._protocol = TemperatureProtocol(self._connection)
+        self.connection = IPConnection(settings)
+        self._protocol = TemperatureProtocol(self.connection)
 
         super().__init__()
 
@@ -54,9 +55,6 @@ async def _get_ramp_rate(self) -> float:
     async def _set_ramp_rate(self, value: float) -> None:
         await self._protocol.send_command("R", value, float)
 
-    async def connect(self):
-        await self._connection.connect(self._ip_settings)
-
 
 gui_options = EpicsGUIOptions(output_dir=Path("."), title="Demo Temperature Controller")
 epics_ca = EpicsCATransport(gui=gui_options)
diff --git a/docs/snippets/static10.py b/docs/snippets/static10.py
index e6ea3292d..55f7374cb 100644
--- a/docs/snippets/static10.py
+++ b/docs/snippets/static10.py
@@ -27,8 +27,11 @@ async def send_query(self, param: str, dtype: type[ValueT]) -> ValueT:
 
 
 class TemperatureRampController(Controller):
+    connection: IPConnection
+
     def __init__(self, index: int, connection: IPConnection) -> None:
         suffix = f"{index:02d}"
+        self.connection = connection
         self._protocol = TemperatureProtocol(connection, suffix)
         super().__init__(f"Ramp{suffix}")
 
@@ -53,10 +56,11 @@ async def _set_end(self, value: int) -> None:
 
 
 class TemperatureController(Controller):
+    connection: IPConnection
+
     def __init__(self, ramp_count: int, settings: IPConnectionSettings):
-        self._ip_settings = settings
-        self._connection = IPConnection()
-        self._protocol = TemperatureProtocol(self._connection)
+        self.connection = IPConnection(settings)
+        self._protocol = TemperatureProtocol(self.connection)
 
         super().__init__()
 
@@ -70,7 +74,7 @@ def __init__(self, ramp_count: int, settings: IPConnectionSettings):
 
         self._ramp_controllers: list[TemperatureRampController] = []
         for index in range(1, ramp_count + 1):
-            controller = TemperatureRampController(index, self._connection)
+            controller = TemperatureRampController(index, self.connection)
             self._ramp_controllers.append(controller)
             self.add_sub_controller(f"R{index}", controller)
 
@@ -86,9 +90,6 @@ async def _get_ramp_rate(self) -> float:
     async def _set_ramp_rate(self, value: float) -> None:
         await self._protocol.send_command("R", value, float)
 
-    async def connect(self):
-        await self._connection.connect(self._ip_settings)
-
 
 gui_options = EpicsGUIOptions(output_dir=Path("."), title="Demo Temperature Controller")
 epics_ca = EpicsCATransport(gui=gui_options)
diff --git a/docs/snippets/static11.py b/docs/snippets/static11.py
index 222d0cdda..ff6664398 100644
--- a/docs/snippets/static11.py
+++ b/docs/snippets/static11.py
@@ -33,8 +33,11 @@ class OnOffEnum(enum.StrEnum):
 
 
 class TemperatureRampController(Controller):
+    connection: IPConnection
+
     def __init__(self, index: int, connection: IPConnection) -> None:
         suffix = f"{index:02d}"
+        self.connection = connection
         self._protocol = TemperatureProtocol(connection, suffix)
         super().__init__(f"Ramp{suffix}")
 
@@ -70,10 +73,11 @@ async def _set_enabled(self, value: OnOffEnum) -> None:
 
 
 class TemperatureController(Controller):
+    connection: IPConnection
+
     def __init__(self, ramp_count: int, settings: IPConnectionSettings):
-        self._ip_settings = settings
-        self._connection = IPConnection()
-        self._protocol = TemperatureProtocol(self._connection)
+        self.connection = IPConnection(settings)
+        self._protocol = TemperatureProtocol(self.connection)
 
         super().__init__()
 
@@ -87,7 +91,7 @@ def __init__(self, ramp_count: int, settings: IPConnectionSettings):
 
         self._ramp_controllers: list[TemperatureRampController] = []
         for index in range(1, ramp_count + 1):
-            controller = TemperatureRampController(index, self._connection)
+            controller = TemperatureRampController(index, self.connection)
             self._ramp_controllers.append(controller)
             self.add_sub_controller(f"R{index}", controller)
 
@@ -103,9 +107,6 @@ async def _get_ramp_rate(self) -> float:
     async def _set_ramp_rate(self, value: float) -> None:
         await self._protocol.send_command("R", value, float)
 
-    async def connect(self):
-        await self._connection.connect(self._ip_settings)
-
 
 gui_options = EpicsGUIOptions(output_dir=Path("."), title="Demo Temperature Controller")
 epics_ca = EpicsCATransport(gui=gui_options)
diff --git a/docs/snippets/static12.py b/docs/snippets/static12.py
index db7ced42c..267d345d5 100644
--- a/docs/snippets/static12.py
+++ b/docs/snippets/static12.py
@@ -35,8 +35,11 @@ class OnOffEnum(enum.StrEnum):
 
 
 class TemperatureRampController(Controller):
+    connection: IPConnection
+
     def __init__(self, index: int, connection: IPConnection) -> None:
         suffix = f"{index:02d}"
+        self.connection = connection
         self._protocol = TemperatureProtocol(connection, suffix)
         super().__init__(f"Ramp{suffix}")
 
@@ -81,10 +84,11 @@ async def _get_actual(self) -> float:
 
 
 class TemperatureController(Controller):
+    connection: IPConnection
+
     def __init__(self, ramp_count: int, settings: IPConnectionSettings):
-        self._ip_settings = settings
-        self._connection = IPConnection()
-        self._protocol = TemperatureProtocol(self._connection)
+        self.connection = IPConnection(settings)
+        self._protocol = TemperatureProtocol(self.connection)
 
         super().__init__()
 
@@ -98,7 +102,7 @@ def __init__(self, ramp_count: int, settings: IPConnectionSettings):
 
         self._ramp_controllers: list[TemperatureRampController] = []
         for index in range(1, ramp_count + 1):
-            controller = TemperatureRampController(index, self._connection)
+            controller = TemperatureRampController(index, self.connection)
             self._ramp_controllers.append(controller)
             self.add_sub_controller(f"R{index}", controller)
 
@@ -114,13 +118,10 @@ async def _get_ramp_rate(self) -> float:
     async def _set_ramp_rate(self, value: float) -> None:
         await self._protocol.send_command("R", value, float)
 
-    async def connect(self):
-        await self._connection.connect(self._ip_settings)
-
     @scan(0.1)
     async def update_voltages(self):
         voltages = json.loads(
-            (await self._connection.send_query("V?\r\n")).strip("\r\n")
+            (await self.connection.send_query("V?\r\n")).strip("\r\n")
         )
         for index, controller in enumerate(self._ramp_controllers):
             await controller.voltage.update(float(voltages[index]))
diff --git a/docs/snippets/static13.py b/docs/snippets/static13.py
index 420acd994..717ca503a 100644
--- a/docs/snippets/static13.py
+++ b/docs/snippets/static13.py
@@ -36,8 +36,11 @@ class OnOffEnum(enum.StrEnum):
 
 
 class TemperatureRampController(Controller):
+    connection: IPConnection
+
     def __init__(self, index: int, connection: IPConnection) -> None:
         suffix = f"{index:02d}"
+        self.connection = connection
         self._protocol = TemperatureProtocol(connection, suffix)
         super().__init__(f"Ramp{suffix}")
 
@@ -82,10 +85,11 @@ async def _get_actual(self) -> float:
 
 
 class TemperatureController(Controller):
+    connection: IPConnection
+
     def __init__(self, ramp_count: int, settings: IPConnectionSettings):
-        self._ip_settings = settings
-        self._connection = IPConnection()
-        self._protocol = TemperatureProtocol(self._connection)
+        self.connection = IPConnection(settings)
+        self._protocol = TemperatureProtocol(self.connection)
 
         super().__init__()
 
@@ -99,7 +103,7 @@ def __init__(self, ramp_count: int, settings: IPConnectionSettings):
 
         self._ramp_controllers: list[TemperatureRampController] = []
         for index in range(1, ramp_count + 1):
-            controller = TemperatureRampController(index, self._connection)
+            controller = TemperatureRampController(index, self.connection)
             self._ramp_controllers.append(controller)
             self.add_sub_controller(f"R{index}", controller)
 
@@ -115,13 +119,10 @@ async def _get_ramp_rate(self) -> float:
     async def _set_ramp_rate(self, value: float) -> None:
         await self._protocol.send_command("R", value, float)
 
-    async def connect(self):
-        await self._connection.connect(self._ip_settings)
-
     @scan(0.1)
     async def update_voltages(self):
         voltages = json.loads(
-            (await self._connection.send_query("V?\r\n")).strip("\r\n")
+            (await self.connection.send_query("V?\r\n")).strip("\r\n")
         )
         for index, controller in enumerate(self._ramp_controllers):
             await controller.voltage.update(float(voltages[index]))
diff --git a/docs/snippets/static14.py b/docs/snippets/static14.py
index 9e25a6418..50ead46ae 100644
--- a/docs/snippets/static14.py
+++ b/docs/snippets/static14.py
@@ -40,8 +40,11 @@ class OnOffEnum(enum.StrEnum):
 
 
 class TemperatureRampController(Controller):
+    connection: IPConnection
+
     def __init__(self, index: int, connection: IPConnection) -> None:
         suffix = f"{index:02d}"
+        self.connection = connection
         self._protocol = TemperatureProtocol(connection, suffix)
         super().__init__(f"Ramp{suffix}")
 
@@ -86,10 +89,11 @@ async def _get_actual(self) -> float:
 
 
 class TemperatureController(Controller):
+    connection: IPConnection
+
     def __init__(self, ramp_count: int, settings: IPConnectionSettings):
-        self._ip_settings = settings
-        self._connection = IPConnection()
-        self._protocol = TemperatureProtocol(self._connection)
+        self.connection = IPConnection(settings)
+        self._protocol = TemperatureProtocol(self.connection)
 
         super().__init__()
 
@@ -103,7 +107,7 @@ def __init__(self, ramp_count: int, settings: IPConnectionSettings):
 
         self._ramp_controllers: list[TemperatureRampController] = []
         for index in range(1, ramp_count + 1):
-            controller = TemperatureRampController(index, self._connection)
+            controller = TemperatureRampController(index, self.connection)
             self._ramp_controllers.append(controller)
             self.add_sub_controller(f"R{index}", controller)
 
@@ -119,13 +123,10 @@ async def _get_ramp_rate(self) -> float:
     async def _set_ramp_rate(self, value: float) -> None:
         await self._protocol.send_command("R", value, float)
 
-    async def connect(self):
-        await self._connection.connect(self._ip_settings)
-
     @scan(0.1)
     async def update_voltages(self):
         voltages = json.loads(
-            (await self._connection.send_query("V?\r\n")).strip("\r\n")
+            (await self.connection.send_query("V?\r\n")).strip("\r\n")
         )
         for index, controller in enumerate(self._ramp_controllers):
             await controller.voltage.update(float(voltages[index]))
diff --git a/docs/snippets/static15.py b/docs/snippets/static15.py
index ac1a1d0d9..b7b932cbb 100644
--- a/docs/snippets/static15.py
+++ b/docs/snippets/static15.py
@@ -48,8 +48,11 @@ class OnOffEnum(enum.StrEnum):
 
 
 class TemperatureRampController(Controller):
+    connection: IPConnection
+
     def __init__(self, index: int, connection: IPConnection) -> None:
         suffix = f"{index:02d}"
+        self.connection = connection
         self._protocol = TemperatureProtocol(connection, suffix)
         super().__init__(f"Ramp{suffix}")
 
@@ -94,10 +97,11 @@ async def _get_actual(self) -> float:
 
 
 class TemperatureController(Controller):
+    connection: IPConnection
+
     def __init__(self, ramp_count: int, settings: IPConnectionSettings):
-        self._ip_settings = settings
-        self._connection = IPConnection()
-        self._protocol = TemperatureProtocol(self._connection)
+        self.connection = IPConnection(settings)
+        self._protocol = TemperatureProtocol(self.connection)
 
         super().__init__()
 
@@ -111,7 +115,7 @@ def __init__(self, ramp_count: int, settings: IPConnectionSettings):
 
         self._ramp_controllers: list[TemperatureRampController] = []
         for index in range(1, ramp_count + 1):
-            controller = TemperatureRampController(index, self._connection)
+            controller = TemperatureRampController(index, self.connection)
             self._ramp_controllers.append(controller)
             self.add_sub_controller(f"R{index}", controller)
 
@@ -127,13 +131,10 @@ async def _get_ramp_rate(self) -> float:
     async def _set_ramp_rate(self, value: float) -> None:
         await self._protocol.send_command("R", value, float)
 
-    async def connect(self):
-        await self._connection.connect(self._ip_settings)
-
     @scan(0.1)
     async def update_voltages(self):
         voltages = json.loads(
-            (await self._connection.send_query("V?\r\n")).strip("\r\n")
+            (await self.connection.send_query("V?\r\n")).strip("\r\n")
         )
         for index, controller in enumerate(self._ramp_controllers):
             await controller.voltage.update(float(voltages[index]))
diff --git a/docs/tutorials/dynamic-drivers.md b/docs/tutorials/dynamic-drivers.md
index ae55dc608..84a6552b8 100644
--- a/docs/tutorials/dynamic-drivers.md
+++ b/docs/tutorials/dynamic-drivers.md
@@ -31,14 +31,14 @@ or a description for the parameter.
 
 ## FastCS Initialisation
 
-Specific `Controller` classes can optionally implement an async `initialise` method to
+Specific `Controller` classes can optionally implement an async `build` method to
 perform any start up logic. The intention here is that the `__init__` method should be
-minimal and the `initialise` method performs any long running calls, such as querying an
+minimal and the `build` method performs any long running calls, such as querying an
 API, allowing FastCS to run these concurrently to reduce start times.
 
 Take the driver implementation from the previous tutorial and remove the
 statically defined `Attributes` and creation of sub controllers in `__init__`. Then
-implement an `initialise` method to create these dynamically instead.
+implement a `build` method to create these dynamically instead.
 
 Create a pydantic model to validate the response from the device
 
@@ -56,12 +56,12 @@ construction time just like statically-declared ones do.
 :lines: 50-79
 :::
 
-Update the controllers to not define attributes statically and implement initialise
+Update the controllers to not define attributes statically and implement build
 methods to create these attributes dynamically, passing the shared `TemperatureProtocol`
 down to `create_attributes` so the dynamically-created getters/setters can use it.
 
 :::{literalinclude} /snippets/dynamic.py
-:lines: 82-128
+:lines: 82-131
 :::
 
 TODO: Add `enabled` back in to `TemperatureRampController` and recreate `disable_all` to
diff --git a/docs/tutorials/static-drivers.md b/docs/tutorials/static-drivers.md
index e90248e9f..b654ea375 100644
--- a/docs/tutorials/static-drivers.md
+++ b/docs/tutorials/static-drivers.md
@@ -145,9 +145,10 @@ its own getter/setter logic and connection, but there are some built in connecti
 options.
 
 Update the controller to create an `IPConnection` to communicate with the simulator over
-TCP and implement a `connect` method that establishes the connection. The `connect`
-method is called by the FastCS application at the appropriate time during start up to
-ensure the connection is established before it is used.
+TCP, giving it the settings it needs. A driver never opens the connection itself: FastCS
+opens it at the appropriate time during start up, before anything uses it, and reopens
+it if it drops. Declaring `connection: IPConnection` on the class narrows the base
+class's connection so this controller's own code can call `IPConnection`'s methods.
 
 :::{note}
 The simulator control connection is on port 25565.
@@ -157,7 +158,7 @@ The simulator control connection is on port 25565.
 :class: dropdown, hint
 
 :::{literalinclude} /snippets/static06.py
-:emphasize-lines: 4,15-22,27-28
+:emphasize-lines: 4,12,19,25-26
 :::
 
 ::::
@@ -183,7 +184,7 @@ Passing the getter bare, as here, means it is called once at start up. Wrap it i
 :class: dropdown, hint
 
 :::{literalinclude} /snippets/static07.py
-:emphasize-lines: 13-19,21-23
+:emphasize-lines: 12,15,19,21-23
 :::
 
 ::::
@@ -226,7 +227,7 @@ constructor to perform the cast.
 :class: dropdown, hint
 
 :::{literalinclude} /snippets/static08.py
-:emphasize-lines: 12,15-27,34,38-39,41-45
+:emphasize-lines: 12,15-27,35,39-40,42-46
 :::
 
 ::::
@@ -250,7 +251,7 @@ The set commands do not return a response, so the setter uses `send_command` ins
 :class: dropdown, hint
 
 :::{literalinclude} /snippets/static09.py
-:emphasize-lines: 4,40-45,53-57
+:emphasize-lines: 4,41-46,54-57
 :::
 
 ::::
@@ -291,11 +292,15 @@ Create a `TemperatureRampController` with two `AttrRW`s for the ramp start and e
 to define how many ramps there are, which is used to register the correct number of ramp
 controllers with the parent.
 
+Each ramp holds the *same* `IPConnection` object as its parent rather than one of its
+own, so the whole tree has one health state and one reconnect task between it: when the
+link drops, every ramp's polling pauses with it and they all resume together.
+
 ::::{admonition} Code 10
 :class: dropdown, hint
 
 :::{literalinclude} /snippets/static10.py
-:emphasize-lines: 30-53,57,73-77
+:emphasize-lines: 30,32-56,75-79
 :::
 
 ::::
@@ -320,7 +325,7 @@ Add an `AttrRW` to the `TemperatureRampController`s with an `Enum` type, using a
 :class: dropdown, hint
 
 :::{literalinclude} /snippets/static11.py
-:emphasize-lines: 1,31-33,48-53,67-71
+:emphasize-lines: 1,31-33,51-56,70-74
 :::
 
 ::::
@@ -375,7 +380,7 @@ above.
 :class: dropdown, hint
 
 :::{literalinclude} /snippets/static12.py
-:emphasize-lines: 11,56-58,78-82,123-129
+:emphasize-lines: 11,59-61,81-85,121-127
 :::
 
 ::::
@@ -396,7 +401,7 @@ controller by calling `set` on each `enabled` attribute.
 :class: dropdown, hint
 
 :::{literalinclude} /snippets/static13.py
-:emphasize-lines: 1,132-137
+:emphasize-lines: 1,133-138
 :::
 
 ::::
@@ -428,7 +433,7 @@ inside `TemperatureProtocol.send_command` to log the commands it sends.
 :class: dropdown, hint
 
 :::{literalinclude} /snippets/static14.py
-:emphasize-lines: 12,28,145,150
+:emphasize-lines: 12,28,146,151
 :::
 
 ::::
@@ -456,7 +461,7 @@ is enabled the messages are visible.
 :class: dropdown, hint
 
 :::{literalinclude} /snippets/static15.py
-:emphasize-lines: 12,14,21,34-36,41,125,153
+:emphasize-lines: 12,14,21,34-36,41,129,154
 :::
 
 ::::
diff --git a/src/fastcs/connections/__init__.py b/src/fastcs/connections/__init__.py
index 5001409a1..a733ac1ad 100644
--- a/src/fastcs/connections/__init__.py
+++ b/src/fastcs/connections/__init__.py
@@ -1,5 +1,9 @@
+from .connection import DEFAULT_MAX_ATTEMPTS as DEFAULT_MAX_ATTEMPTS
+from .connection import DEFAULT_RECONNECT_PERIOD as DEFAULT_RECONNECT_PERIOD
+from .connection import Connection as Connection
 from .ip_connection import IPConnection as IPConnection
 from .ip_connection import IPConnectionSettings as IPConnectionSettings
 from .ip_connection import StreamConnection as StreamConnection
+from .registry import Connections as Connections
 from .serial_connection import SerialConnection as SerialConnection
 from .serial_connection import SerialConnectionSettings as SerialConnectionSettings
diff --git a/src/fastcs/connections/connection.py b/src/fastcs/connections/connection.py
new file mode 100644
index 000000000..3c5fc1528
--- /dev/null
+++ b/src/fastcs/connections/connection.py
@@ -0,0 +1,133 @@
+from __future__ import annotations
+
+import asyncio
+from abc import ABC, abstractmethod
+from typing import Generic, TypeVar
+
+Introspection_T = TypeVar("Introspection_T")
+
+DEFAULT_RECONNECT_PERIOD = 1.0
+"""Seconds a connection waits between reconnect attempts, unless it says otherwise."""
+
+DEFAULT_MAX_ATTEMPTS = 10
+"""Reconnect attempts a connection makes before giving up, unless it says otherwise."""
+
+
+class Connection(ABC, Generic[Introspection_T]):
+    """A link to hardware. Owns its own health state.
+
+    Several controllers may share one instance - a sub controller that talks to the
+    same device as its parent holds the same object rather than consulting the parent.
+    Failure, gating and recovery all resolve through that shared object, so a tree of
+    controllers behind one socket has one health state, one reconnect task and one
+    retry budget between them.
+
+    A concrete connection opens the link in ``connect`` and closes it in ``close``,
+    and calls `set_disconnected` from its own IO when the *transport* fails. That is the
+    one place that can tell "the socket died" from "the device rejected that
+    parameter", and only the first is a connection failure::
+
+        async def get(self, path: str):
+            try:
+                response = await self._client.get(path)
+            except (ConnectError, ReadTimeout):
+                self.set_disconnected()  # transport is gone
+                raise
+            response.raise_for_status()  # a device complaint, not a dead link
+            return response.json()["value"]
+
+    Nothing above a connection has to catch anything, and no exception type is a
+    contract between layers.
+
+    The `ControllerRunner` keys its per-connection state by identity, so a
+    ``Connection`` must never define ``__eq__``: two sockets with matching settings
+    are two connections, and an ``__eq__`` would silently collapse them.
+
+    Args:
+        depends_on: A connection this one is layered over, if any. Declared, never
+            derived - the runner will not attempt this one while that one is down.
+        reconnect_period: Seconds between reconnect attempts. Defaults to the class
+            attribute of the same name.
+        max_attempts: Consecutive failed attempts before this connection gives up.
+            Defaults to the class attribute of the same name.
+
+    """
+
+    # Class defaults. Framework defaults below, class attributes on a concrete
+    # connection, constructor arguments on top - three tiers, each overriding the last.
+    reconnect_period: float = DEFAULT_RECONNECT_PERIOD
+    max_attempts: int = DEFAULT_MAX_ATTEMPTS
+
+    def __init__(
+        self,
+        depends_on: Connection | None = None,
+        reconnect_period: float | None = None,
+        max_attempts: int | None = None,
+    ) -> None:
+        self._connected = False
+        self._up = asyncio.Event()
+        self._down = asyncio.Event()
+        self._down.set()
+
+        # Declared, never derived. A connection layered over another names it here;
+        # the runner will not attempt this one while that one is down.
+        self.depends_on = depends_on
+
+        if reconnect_period is not None:
+            self.reconnect_period = reconnect_period
+        if max_attempts is not None:
+            self.max_attempts = max_attempts
+
+    @property
+    def connected(self) -> bool:
+        """Whether the link is currently believed to be usable.
+
+        Set by the framework - a driver never writes it. ``connect`` returning cleanly
+        marks it up; `set_disconnected` from the connection's own IO marks it down.
+        """
+        return self._connected
+
+    @abstractmethod
+    async def connect(self) -> Introspection_T:
+        """Open the link, or raise. Return whatever introspection the caller needs.
+
+        This means "make the link usable", not merely "open the socket" - a device
+        that needs a mode set before it can be introspected has that write here,
+        rather than in a controller's ``build``.
+
+        The framework marks the connection connected when this returns cleanly, and
+        compares the return value against the startup value on every reconnect.
+        """
+
+    @abstractmethod
+    async def close(self) -> None:
+        """Close the link. Called at shutdown and before every reconnect attempt.
+
+        Must tolerate being called on a link that is already closed.
+        """
+
+    def set_disconnected(self) -> None:
+        """Called by the connection's own IO when its transport fails.
+
+        Wakes this connection's reconnect task and gates every scan that uses it.
+        """
+        self._connected = False
+        self._up.clear()
+        self._down.set()
+
+    def _set_connected(self) -> None:
+        """Framework only. Wakes anything awaiting this connection's recovery."""
+        self._connected = True
+        self._down.clear()
+        self._up.set()
+
+    async def wait_up(self) -> None:
+        """Block until this connection is up. Returns immediately if it already is."""
+        await self._up.wait()
+
+    async def wait_down(self) -> None:
+        """Block until this connection is down. Returns immediately if it already is."""
+        await self._down.wait()
+
+    def __repr__(self) -> str:
+        return f"{type(self).__name__}(connected={self._connected})"
diff --git a/src/fastcs/connections/ip_connection.py b/src/fastcs/connections/ip_connection.py
index f021aa913..d8c34e24a 100644
--- a/src/fastcs/connections/ip_connection.py
+++ b/src/fastcs/connections/ip_connection.py
@@ -1,11 +1,16 @@
 import asyncio
 from dataclasses import dataclass
 
+from fastcs.connections.connection import Connection
 from fastcs.tracer import Tracer
 
 
-class DisconnectedError(Exception):
-    """Raised if the ip connection is disconnected."""
+class DisconnectedError(ConnectionError):
+    """Raised if the ip connection is disconnected.
+
+    A `ConnectionError`, and so an `OSError`, because that is what the rest of this
+    module treats as "the transport is gone" rather than "the device complained".
+    """
 
     pass
 
@@ -46,12 +51,26 @@ async def close(self):
         await self.writer.wait_closed()
 
 
-class IPConnection(Tracer):
-    """For connecting to an ip using a `StreamConnection`."""
+class IPConnection(Connection[None], Tracer):
+    """For connecting to an ip using a `StreamConnection`.
+
+    The settings are given at construction rather than to ``connect``, because the
+    framework opens and reopens the link without knowing anything about it. IO
+    marks the connection down when the *transport* fails, so everything holding it
+    stops and its reconnect task wakes.
+
+    Args:
+        settings: Where to connect to
+        kwargs: Passed to `Connection` - ``depends_on``, ``reconnect_period``,
+            ``max_attempts``
+
+    """
 
-    def __init__(self):
-        super().__init__()
-        self.__connection = None
+    def __init__(self, settings: IPConnectionSettings | None = None, **kwargs) -> None:
+        Connection.__init__(self, **kwargs)
+        Tracer.__init__(self)
+        self._settings = settings or IPConnectionSettings()
+        self.__connection: StreamConnection | None = None
 
     @property
     def _connection(self) -> StreamConnection:
@@ -60,18 +79,40 @@ def _connection(self) -> StreamConnection:
 
         return self.__connection
 
-    async def connect(self, settings: IPConnectionSettings):
-        reader, writer = await asyncio.open_connection(settings.ip, settings.port)
+    async def connect(self) -> None:
+        reader, writer = await asyncio.open_connection(
+            self._settings.ip, self._settings.port
+        )
         self.__connection = StreamConnection(reader, writer)
 
     async def send_command(self, message: str) -> None:
         async with self._connection as connection:
-            await connection.send_message(message)
+            try:
+                await connection.send_message(message)
+            except OSError:
+                # The socket is gone, rather than the device complaining. Everything
+                # holding this connection is now down.
+                self.set_disconnected()
+                raise
 
     async def send_query(self, message: str) -> str:
         async with self._connection as connection:
-            await connection.send_message(message)
-            response = await connection.receive_response()
+            try:
+                await connection.send_message(message)
+                response = await connection.receive_response()
+                if not response:
+                    # ``readline`` returns b"" at EOF, so a peer that closed the
+                    # socket rather than answering looks like an empty reply. It
+                    # is a dead link, and nothing else here would notice: the
+                    # caller would get "" and fail to parse it, over and over,
+                    # while the reconnect task stayed idle.
+                    raise DisconnectedError(
+                        "Connection closed by peer while awaiting a response"
+                    )
+            except OSError:
+                self.set_disconnected()
+                raise
+
             self.log_event(
                 "Received query response",
                 query=message.strip(),
@@ -79,7 +120,7 @@ async def send_query(self, message: str) -> str:
             )
             return response
 
-    async def close(self):
+    async def close(self) -> None:
         if self.__connection is None:
             return
 
diff --git a/src/fastcs/connections/registry.py b/src/fastcs/connections/registry.py
new file mode 100644
index 000000000..cc5de2838
--- /dev/null
+++ b/src/fastcs/connections/registry.py
@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+from typing import TypeVar
+
+from fastcs.connections.connection import Connection
+
+Connection_T = TypeVar("Connection_T", bound=Connection)
+
+
+class Connections:
+    """The connections available to a controller tree, claimed by name.
+
+    Built once - by the launcher from the ``connections:`` block, or by hand - and
+    forwarded down the tree. A controller claims what it needs with `get` rather than
+    receiving a bare connection object, so a controller's constructor signature does
+    not change when something three tiers below it needs a new connection::
+
+        class EigerController(Controller):
+            def __init__(self, connections: Connections) -> None:
+                super().__init__()
+                self.add_sub_controller("DET", EigerDetectorController(connections))
+                self.add_sub_controller("OD", OdinController(connections))
+
+        class EigerDetectorController(Controller):
+            def __init__(self, connections: Connections) -> None:
+                self.connection = connections.get("eiger", EigerConnection)
+                super().__init__()
+
+    Args:
+        connections: The declared connections, keyed by the name controllers claim
+            them under. Iteration order is declaration order, which is the order the
+            runner opens them in.
+
+    """
+
+    def __init__(self, connections: dict[str, Connection]) -> None:
+        self._connections = dict(connections)
+        self._claimed: set[str] = set()
+
+    def get(self, name: str, expected: type[Connection_T]) -> Connection_T:
+        """Claim a connection by name, asserting its type.
+
+        Called from ``__init__``, so a bad name or type fails at construction -
+        before anything is opened - rather than at the first IO.
+
+        Args:
+            name: The name the connection was declared under
+            expected: The `Connection` subclass the caller intends to use
+
+        Returns:
+            The declared connection
+
+        Raises:
+            KeyError: If nothing was declared under that name
+            TypeError: If what was declared is not an ``expected``
+
+        """
+        try:
+            connection = self._connections[name]
+        except KeyError:
+            raise KeyError(
+                f"No connection named {name!r}. Declared: {sorted(self._connections)}"
+            ) from None
+
+        if not isinstance(connection, expected):
+            raise TypeError(
+                f"Connection {name!r} is {type(connection).__name__}, "
+                f"but {expected.__name__} was expected"
+            )
+
+        self._claimed.add(name)
+        return connection
+
+    def unclaimed(self) -> set[str]:
+        """Names declared but never claimed.
+
+        A config typo that would otherwise be opened and reconnected forever while
+        doing nothing, so the runner warns about it at startup.
+        """
+        return set(self._connections) - self._claimed
+
+    def name_of(self, connection: Connection) -> str | None:
+        """The name a connection was declared under, by identity."""
+        for name, declared in self._connections.items():
+            if declared is connection:
+                return name
+        return None
+
+    def values(self) -> list[Connection]:
+        """The declared connections, in declaration order."""
+        return list(self._connections.values())
+
+    def __contains__(self, name: object) -> bool:
+        return name in self._connections
+
+    def __len__(self) -> int:
+        return len(self._connections)
+
+    def __repr__(self) -> str:
+        return f"Connections({sorted(self._connections)})"
diff --git a/src/fastcs/connections/serial_connection.py b/src/fastcs/connections/serial_connection.py
index 65bbf6801..4d3f1c82b 100644
--- a/src/fastcs/connections/serial_connection.py
+++ b/src/fastcs/connections/serial_connection.py
@@ -3,6 +3,8 @@
 
 import aioserial
 
+from fastcs.connections.connection import Connection
+
 
 class NotOpenedError(Exception):
     """If the serial stream is not opened."""
@@ -16,15 +18,29 @@ class SerialConnectionSettings:
     baud: int = 115200
 
 
-class SerialConnection:
-    """A serial connection."""
+class SerialConnection(Connection[None]):
+    """A serial connection.
+
+    The settings are given at construction rather than to ``connect``, because the
+    framework opens and reopens the link without knowing anything about it.
+
+    Args:
+        settings: Which port to open, and at what baud rate
+        kwargs: Passed to `Connection` - ``depends_on``, ``reconnect_period``,
+            ``max_attempts``
 
-    def __init__(self):
-        self.stream = None
+    """
+
+    def __init__(self, settings: SerialConnectionSettings, **kwargs) -> None:
+        super().__init__(**kwargs)
+        self._settings = settings
         self._lock = asyncio.Lock()
+        self.__stream: aioserial.AioSerial | None = None
 
-    async def connect(self, settings: SerialConnectionSettings) -> None:
-        self.__stream = aioserial.AioSerial(port=settings.port, baudrate=settings.baud)
+    async def connect(self) -> None:
+        self.__stream = aioserial.AioSerial(
+            port=self._settings.port, baudrate=self._settings.baud
+        )
 
     @property
     def _stream(self) -> aioserial.AioSerial:
@@ -45,12 +61,24 @@ async def send_query(self, message: bytes, response_size: int) -> bytes:
             return await self._receive_response(response_size)
 
     async def _send_message(self, message):
-        await self._stream.write_async(message)
+        try:
+            await self._stream.write_async(message)
+        except (OSError, aioserial.SerialException):
+            # The port is gone, rather than the device complaining.
+            self.set_disconnected()
+            raise
 
     async def _receive_response(self, size):
-        return await self._stream.read_async(size)
+        try:
+            return await self._stream.read_async(size)
+        except (OSError, aioserial.SerialException):
+            self.set_disconnected()
+            raise
 
     async def close(self) -> None:
         async with self._lock:
-            self._stream.close()
+            if self.__stream is None:
+                return
+
+            self.__stream.close()
             self.__stream = None
diff --git a/src/fastcs/control_system.py b/src/fastcs/control_system.py
index 44467fe63..738712307 100644
--- a/src/fastcs/control_system.py
+++ b/src/fastcs/control_system.py
@@ -95,10 +95,34 @@ async def serve(self, interactive: bool = True) -> None:
             interactive: Whether to create an interactive IPython shell
 
         """
+        try:
+            coros = await self._start(interactive)
+        except BaseException:
+            # A failure during startup aborts: a partly built application is worse
+            # than none, because clients connect successfully and never find what
+            # they are looking for. Close whatever was opened on the way up, then
+            # let the caller - or the orchestrator - see why.
+            await self._runner.stop()
+            raise
+
+        try:
+            await asyncio.gather(*coros)
+        except asyncio.CancelledError:
+            pass
+        except Exception:
+            logger.exception("Unhandled exception in serve")
+        finally:
+            logger.info("Shutting down FastCS")
+            await self._runner.stop()
+            if self._runner.fatal_reason is not None:
+                raise self._runner.fatal_reason
+
+    async def _start(self, interactive: bool) -> list[Coroutine]:
+        """Bring the application up, and return what ``serve`` should await."""
         # Build the APIs before wiring transports to them: a transport
         # registers its callbacks when it connects, and would miss the first
         # readback if the controllers had already started.
-        self.controller_apis = await self._runner.setup()
+        self.controller_apis = await self._runner.build()
 
         context = {
             "controllers": {_context_key(c): c for c in self._controllers},
@@ -143,15 +167,20 @@ async def block_forever():
 
         await self._runner.start()
 
-        try:
-            await asyncio.gather(*coros)
-        except asyncio.CancelledError:
-            pass
-        except Exception:
-            logger.exception("Unhandled exception in serve")
-        finally:
-            logger.info("Shutting down FastCS")
-            await self._runner.stop()
+        # A fatal runner condition - a device coming back describing itself
+        # differently, say - happens in a background task, where a raise would be
+        # invisible. The runner records it instead, and this coroutine is where the
+        # process notices and comes down rather than serving a tree that no longer
+        # matches the hardware. Nothing calls ``sys.exit``, so an embedder sees an
+        # exception out of ``serve`` rather than losing its process.
+        async def fail_on_fatal() -> None:
+            await self._runner.fatal_error.wait()
+            assert self._runner.fatal_reason is not None
+            raise self._runner.fatal_reason
+
+        coros.append(fail_on_fatal())
+
+        return coros
 
     async def _interactive_shell(self, context: dict[str, Any]):
         """Spawn interactive shell in another thread and wait for it to complete."""
diff --git a/src/fastcs/controllers/base_controller.py b/src/fastcs/controllers/base_controller.py
index ef3b76252..350300923 100755
--- a/src/fastcs/controllers/base_controller.py
+++ b/src/fastcs/controllers/base_controller.py
@@ -3,6 +3,7 @@
 from copy import deepcopy
 from inspect import getattr_static
 from typing import (
+    Any,
     TypeVar,
     _GenericAlias,  # type: ignore
     get_args,
@@ -36,6 +37,29 @@ class BaseController(Tracer):
     root_attribute: Attribute | None = None
     description: str | None = None
 
+    connection: Any = None
+    """The link this controller does its IO over, if it has one.
+
+    A `Connection`, or ``None``. Set in ``__init__``, usually by claiming it from a
+    `Connections` registry. A controller holds at most one - two devices means two
+    controllers - and several controllers may hold the same object, in which case
+    they share one health state, one reconnect task and one retry budget.
+
+    A controller with no connection (a soft grouping controller, or a
+    `ControllerVector`) is never gated and never reconnected.
+
+    Typed ``Any`` rather than ``Connection | None`` so that a driver can narrow it to
+    the connection it actually holds, and call that connection's own methods::
+
+        class TemperatureController(Controller):
+            connection: IPConnection
+
+    A mutable attribute is invariant, so a driver cannot narrow a declared
+    ``Connection | None`` without a type checker objecting to every driver in
+    existence. The framework reads this attribute in exactly two places - the scan
+    gate and the runner - both of which state the type they expect.
+    """
+
     def __init__(
         self,
         path: list[str] | None = None,
@@ -179,13 +203,35 @@ def __setattr__(self, name, value):
         else:
             super().__setattr__(name, value)
 
-    async def initialise(self):
-        """Hook for subclasses to dynamically add attributes before building the API"""
+    async def build(self):
+        """Hook for structure that depends on the device.
+
+        Called by the framework once this controller's connection is open, and
+        before anything is set up. Add the attributes and sub controllers that could
+        only be known by asking the device - the ones knowable without it belong in
+        ``__init__``, which is where a controller can be constructed and inspected in
+        a test with no hardware.
+
+        A controller whose connection returns introspection from ``connect`` may
+        declare ``build(self, info)`` to receive it; the framework inspects the
+        signature, so ``build(self)`` is equally valid.
+
+        No hardware *writes* here. A device that needs a mode set before
+        introspection works has that write in `Connection.connect`, which means
+        "make the link usable" rather than merely "open the socket".
+        """
         pass
 
-    def post_initialise(self):
-        """Hook to call after all attributes added, before serving the application"""
-        self._validate_type_hints()
+    async def setup(self):
+        """Hook for hardware writes and checks, once the whole tree is built.
+
+        Called by the framework after every controller's ``build`` has run and every
+        connection is open, so this can read and write across the tree.
+
+        No new attributes or sub controllers here - anything created now would never
+        get its own ``build`` or ``setup`` called.
+        """
+        pass
 
     def _validate_type_hints(self):
         """Validate all type-hints were introspected"""
diff --git a/src/fastcs/controllers/controller.py b/src/fastcs/controllers/controller.py
index 0bee8d7d8..781a5438e 100755
--- a/src/fastcs/controllers/controller.py
+++ b/src/fastcs/controllers/controller.py
@@ -3,6 +3,7 @@
 from collections.abc import Sequence
 
 from fastcs.attributes.attr_r import AttrR
+from fastcs.connections import Connection
 from fastcs.controllers.base_controller import BaseController
 from fastcs.controllers.controller_api import ControllerAPI
 from fastcs.logging import logger
@@ -18,7 +19,6 @@ def __init__(
         description: str | None = None,
     ) -> None:
         super().__init__(description=description)
-        self._connected = False
 
     def add_sub_controller(self, name: str, sub_controller: BaseController):
         if name.isdigit():
@@ -30,37 +30,15 @@ def add_sub_controller(self, name: str, sub_controller: BaseController):
 
     @property
     def connected(self) -> bool:
-        """Whether the controller believes it can talk to its device.
+        """Whether this controller can talk to its device.
 
-        Set by `connect`/`reconnect`, and cleared when a scan task raises. The
-        `ControllerRunner` reads it to decide when to reconnect.
+        A read-through to ``self.connection.connected`` - the connection is the one
+        object that knows, and controllers sharing a connection all report the same
+        value, which is correct because they share one link. A controller with no
+        connection has nothing to read through to and is always ``True``.
         """
-        return self._connected
-
-    async def connect(self) -> None:
-        """Hook to perform initial connection to device
-
-        This should set ``_connected`` to ``True`` if the connection was successful to
-        enable scan tasks.
-
-        """
-        self._connected = True
-
-    async def reconnect(self):
-        """Hook to reconnect to device after an error
-
-        This should set ``_connected`` to ``True`` if the connection was successful to
-        enable scan tasks.
-
-        If the connection cannot be re-established it should log an error with the
-        reason. It should not raise an exception.
-
-        """
-        self._connected = True
-
-    async def disconnect(self) -> None:
-        """Hook to tidy up resources before stopping the application"""
-        pass
+        connection: Connection | None = self.connection
+        return connection is None or connection.connected
 
     def create_api_and_tasks(
         self,
@@ -113,10 +91,15 @@ def _create_periodic_scan_coro(
     ) -> ScanCallback:
         """Create a coroutine to run scans at a given period
 
-        This returns a coroutine that runs scans at a given period. If an exception is
-        raised in a callback it is caught and the updates for the controller are
-        paused, waiting for `_connected` to be set back to true via the `reconnect`
-        method.
+        The returned coroutine is gated on this controller's connection: while that
+        connection is down it waits for the connection to come back rather than
+        polling a link that cannot answer. A controller with no connection is never
+        gated.
+
+        The gate is the only thing a scan does about connection health. Failure is
+        detected in exactly one place - the connection's own IO, which knows a dead
+        transport from a device complaint - so a raising scan is logged and retried
+        rather than being read as a disconnection here.
 
         Args:
             period: The period to run the scans at
@@ -128,8 +111,9 @@ def _create_periodic_scan_coro(
 
         async def scan_coro() -> None:
             while True:
-                if not self._connected:
-                    await asyncio.sleep(1)
+                connection: Connection | None = self.connection
+                if connection is not None and not connection.connected:
+                    await connection.wait_up()
                     continue
 
                 try:
@@ -138,9 +122,8 @@ async def scan_coro() -> None:
                     )
                 except Exception:
                     logger.exception("Exception in scan task", period=period)
-                    self._connected = False
-
-                    await asyncio.sleep(1)  # Wait so this message appears last
-                    logger.error("Pausing scan tasks and waiting for reconnect")
+                    # Do not spin: a scan that raises immediately would otherwise
+                    # retry as fast as the event loop allows.
+                    await asyncio.sleep(period)
 
         return scan_coro
diff --git a/src/fastcs/controllers/runner.py b/src/fastcs/controllers/runner.py
index ead653e9e..9ef853ad4 100644
--- a/src/fastcs/controllers/runner.py
+++ b/src/fastcs/controllers/runner.py
@@ -1,41 +1,94 @@
 import asyncio
-from collections.abc import Sequence
+import inspect
+from collections import deque
+from collections.abc import Iterator, Sequence
+from dataclasses import dataclass, field
 
+from fastcs.connections import Connection, Connections
+from fastcs.controllers.base_controller import BaseController
 from fastcs.controllers.controller import Controller
 from fastcs.controllers.controller_api import ControllerAPI
 from fastcs.logging import logger
 from fastcs.methods import ScanCallback
+from fastcs.util import ONCE
 
-RECONNECT_PERIOD = 1.0
-"""Seconds between checks for a controller that has dropped its connection"""
+MAX_BUILD_PASSES = 32
+"""Passes the build phase makes before deciding the tree is not settling.
+
+A ``build`` that adds a sub controller whose ``build`` adds another needs one pass
+per tier; a cap catches runaway construction rather than hanging.
+"""
+
+
+class IntrospectionMismatchError(RuntimeError):
+    """A device came back from a reconnect describing itself differently.
+
+    ``build`` runs once, so there is no way to accommodate the new shape: the
+    application has an attribute tree that no longer matches the hardware. The runner
+    treats this as fatal - it stops, and `ControllerRunner.fatal_error` is set so
+    whatever is running it can exit.
+    """
+
+
+@dataclass
+class _ReconnectState:
+    """What the runner remembers about one connection."""
+
+    introspection: object = None
+    """What ``connect`` returned at startup, compared against on every reconnect."""
+
+    attempts: int = 0
+    """Consecutive failed attempts. Reset by a clean connection."""
+
+    exhausted: asyncio.Event = field(default_factory=asyncio.Event)
+    """Set when this connection has given up. Terminal until the process restarts.
+
+    An `asyncio.Event` rather than a flag because dependents await it: setting it
+    releases anything waiting on this connection, so they stall loudly instead of
+    hanging silently.
+    """
 
 
 class ControllerRunner:
     """Runs one or more `Controller` s, without serving them anywhere.
 
-    This owns the whole controller lifecycle - initialising, connecting,
-    running the initial and periodic tasks, reconnecting after a failure, and
-    tidying up - and nothing about how the controllers are presented. `FastCS`
-    uses it and adds transports on top; an embedded caller that only wants the
-    controllers running can use it on its own::
+    This owns the whole controller lifecycle - opening connections, building and
+    setting up the tree, running the initial and periodic tasks, reconnecting after a
+    failure, and tidying up - and nothing about how the controllers are presented.
+    `FastCS` uses it and adds transports on top; an embedded caller that only wants
+    the controllers running can use it on its own::
 
-        runner = ControllerRunner(controller)
+        runner = ControllerRunner(controller, connections=connections)
         await runner.start()
         ...
         await runner.stop()
 
-    Starting has two halves, because anything serving the controllers needs
-    their `ControllerAPI` before the first values are read: ``setup`` initialises
-    them and builds the APIs, and ``start`` connects and starts the tasks.
-    Calling ``start`` on its own does both.
+    **The runner owns the order of the startup sequence.** Every connection is opened
+    first, then the tree is walked calling ``build``, then ``setup`` runs across the
+    whole built tree, then the tasks start. Controllers never call their own hooks to
+    compensate for sequencing.
+
+    Starting is in two halves, because anything serving the controllers needs their
+    `ControllerAPI` before the first values are read: ``build`` opens the connections,
+    builds the tree and returns the APIs, and ``start`` does the rest. Calling
+    ``start`` on its own does both.
+
+    **A failure anywhere in startup aborts.** A partly built tree means an
+    application with a silently incomplete set of parameters, which is worse than no
+    application at all, because clients connect successfully and never find what they
+    are looking for. The orchestrator owns the retry.
 
-    **Idempotency is the caller's responsibility.** Starting a running runner,
-    or stopping a stopped one, is not defined.
+    **Idempotency is the caller's responsibility.** Starting a running runner, or
+    stopping a stopped one, is not defined.
 
     Args:
         controllers: The controller(s) to run. Accepts either a single
             ``Controller`` or a sequence of them.
         loop: Optional event loop to create the tasks in
+        connections: The declared connections. When given, they are opened in
+            declaration order before the tree is walked, so a ``build`` that adds sub
+            controllers can hand them an already-open connection. When omitted, the
+            runner collects the connections the tree already holds, by identity.
 
     """
 
@@ -43,35 +96,77 @@ def __init__(
         self,
         controllers: Controller | Sequence[Controller],
         loop: asyncio.AbstractEventLoop | None = None,
+        connections: Connections | None = None,
     ) -> None:
         if isinstance(controllers, Controller):
             controllers = [controllers]
         self._controllers: list[Controller] = list(controllers)
         self._loop = loop
+        self._registry = connections
+
+        self._connections: list[Connection] = []
+        self._state: dict[Connection, _ReconnectState] = {}
 
         self._controller_apis: list[ControllerAPI] = []
         self._scan_coros: list[ScanCallback] = []
         self._initial_coros: list[ScanCallback] = []
         self._tasks: set[asyncio.Task] = set()
 
+        self.fatal_error: asyncio.Event = asyncio.Event()
+        """Set when the runner has hit something it cannot carry on from.
+
+        A background task cannot usefully raise - nothing is awaiting it - and an
+        embedded FastCS must not call ``sys.exit``, so a fatal condition is reported
+        here instead. `FastCS` awaits it and shuts down; an embedder can do the same,
+        and read `fatal_reason` for what happened.
+        """
+
+        self.fatal_reason: BaseException | None = None
+        """Why `fatal_error` was set, if it was."""
+
     @property
     def controller_apis(self) -> list[ControllerAPI]:
-        """The API of each controller. Empty until ``setup`` has run."""
+        """The API of each controller. Empty until ``build`` has run."""
         return self._controller_apis
 
-    async def setup(self) -> list[ControllerAPI]:
-        """Initialise the controllers and build their APIs.
+    @property
+    def connections(self) -> list[Connection]:
+        """The connections this runner supervises, in the order it opens them."""
+        return list(self._connections)
+
+    async def build(self) -> list[ControllerAPI]:
+        """Open every connection, build the controller tree and create the APIs.
 
-        Runs before anything connects, so that a transport can be wired to the
-        APIs and catch the first readback.
+        Runs before anything is set up or scanned, so that a transport can be wired
+        to the APIs and catch the first readback.
 
         Returns:
             The API of each controller, in the order they were given
 
         """
+        try:
+            return await self._open_and_build()
+        except BaseException:
+            # Startup aborts, but the connections opened before the failure are
+            # still open, and no task exists yet for a later ``stop`` to be called
+            # to cancel - so nothing else would ever close them.
+            await self._close_connections()
+            raise
+
+    async def _open_and_build(self) -> list[ControllerAPI]:
+        self._connections = self._collect_connections()
+        self._check_dependencies()
+
+        for connection in self._connections:
+            state = _ReconnectState()
+            self._state[connection] = state
+            state.introspection = await connection.connect()
+            connection._set_connected()  # noqa: SLF001
+
+        await self._build_phase()
+
         for controller in self._controllers:
-            await controller.initialise()
-            controller.post_initialise()
+            controller._validate_type_hints()  # noqa: SLF001
 
         self._controller_apis = []
         self._scan_coros = []
@@ -85,15 +180,27 @@ async def setup(self) -> list[ControllerAPI]:
         return self._controller_apis
 
     async def start(self) -> None:
-        """Connect the controllers and start their tasks.
+        """Set the tree up and start its tasks.
 
-        Runs ``setup`` first if it has not already run.
+        Runs ``build`` first if it has not already run.
         """
         if not self._controller_apis:
-            await self.setup()
+            await self.build()
 
-        for controller in self._controllers:
-            await controller.connect()
+        try:
+            await self._setup_and_run()
+        except BaseException:
+            # As in ``build``: a ``setup`` or an initial read that raises leaves
+            # every connection open with nothing to close them.
+            await self.stop()
+            raise
+
+    async def _setup_and_run(self) -> None:
+        for controller in self._walk_controllers():
+            await controller.setup()
+
+        self._warn_about_unclaimed_connections()
+        self._warn_about_unpolled_connections()
 
         for coro in self._initial_coros:
             await coro()
@@ -101,41 +208,388 @@ async def start(self) -> None:
         loop = self._loop or asyncio.get_event_loop()
         self._tasks = {loop.create_task(coro()) for coro in self._scan_coros}
         self._tasks |= {
-            loop.create_task(self._reconnect_loop(controller))
-            for controller in self._controllers
+            loop.create_task(self._reconnect_loop(connection))
+            for connection in self._connections
         }
 
     async def stop(self) -> None:
-        """Stop the tasks and disconnect the controllers."""
+        """Stop the tasks and close every connection.
+
+        Shutdown is a runner operation rather than an author hook: connections are
+        closed in reverse declaration order, so anything layered over another is
+        closed before what it rides on. ``setup`` is not undone - devices keep their
+        last configured state.
+        """
         self._cancel_tasks()
+        await self._close_connections()
 
-        for controller in self._controllers:
+    async def _close_connections(self) -> None:
+        for connection in reversed(self._connections):
             try:
-                await controller.disconnect()
+                await connection.close()
             except Exception:
-                logger.exception(
-                    "Exception during disconnect", controller=controller.path
-                )
+                logger.exception("Exception while closing connection")
+
+    # Startup
 
-    async def _reconnect_loop(self, controller: Controller) -> None:
-        """Bring a controller back after its scan tasks hit an error.
+    def _collect_connections(self) -> list[Connection]:
+        """Every connection the runner supervises, in the order it opens them.
 
-        A scan task that raises marks its controller disconnected and pauses
-        rather than dying, so something has to try to bring it back. That is the
-        runner's job rather than the controller's, so that every controller
-        reconnects the same way whether or not its author thought about it.
+        From the registry when there is one - declaration order, and known before any
+        controller is constructed, which is what lets a ``build`` add a sub controller
+        holding an already-open connection. Otherwise from the tree, level order and
+        deduplicated by identity: two sockets with matching settings are two
+        connections, so identity rather than equality.
         """
-        while True:
-            await asyncio.sleep(RECONNECT_PERIOD)
+        if self._registry is not None:
+            return self._registry.values()
+
+        seen: dict[int, Connection] = {}
+        for controller in self._walk_controllers():
+            connection: Connection | None = controller.connection
+            if connection is not None and id(connection) not in seen:
+                seen[id(connection)] = connection
+        return list(seen.values())
+
+    def _check_dependencies(self) -> None:
+        """``depends_on`` is declared, so it can name anything at all.
+
+        A connection can name one the runner does not supervise, or two can name
+        each other. Either leaves a connection waiting forever with nothing said,
+        so both fail at startup instead.
+        """
+        for connection in self._connections:
+            seen = [connection]
+            dependency = connection.depends_on
+            while dependency is not None:
+                if not self._supervises(dependency):
+                    # It would never be opened, so it would sit at
+                    # ``connected is False`` forever and this connection would
+                    # never be attempted again.
+                    raise ValueError(
+                        f"{type(connection).__name__} depends on a "
+                        f"{type(dependency).__name__} the runner does not "
+                        "supervise, so it would never be opened. Declare it "
+                        "alongside the connection that depends on it."
+                    )
+                if any(dependency is node for node in seen):
+                    chain = " -> ".join(type(node).__name__ for node in seen)
+                    raise ValueError(
+                        f"Cycle in connection dependencies: {chain} -> "
+                        f"{type(dependency).__name__}"
+                    )
+                seen.append(dependency)
+                dependency = dependency.depends_on
+
+    def _supervises(self, connection: Connection) -> bool:
+        """Whether this runner opened, and will reconnect, a connection."""
+        return any(connection is known for known in self._connections)
+
+    async def _build_phase(self) -> None:
+        """Walk the tree top-down calling ``build``, to a fixpoint.
+
+        A ``build`` may add sub controllers, which need building themselves, so the
+        walk repeats over anything newly added until a pass adds nothing.
+        """
+        built: set[int] = set()
+
+        for _ in range(MAX_BUILD_PASSES):
+            pending = [c for c in self._walk_controllers() if id(c) not in built]
+            if not pending:
+                self._check_connections_are_known()
+                return
+
+            for controller in pending:
+                built.add(id(controller))
+                await self._call_build(controller)
+
+        raise RuntimeError(
+            f"Controller tree did not settle in {MAX_BUILD_PASSES} build passes. "
+            "A `build` that adds a sub controller on every pass never finishes."
+        )
 
-            if controller.connected:
+    async def _call_build(self, controller: BaseController) -> None:
+        """Call ``build``, passing the connection's introspection if it wants it.
+
+        ``build(self)`` gets nothing and ``build(self, info)`` gets whatever this
+        controller's connection returned from ``connect``.
+        """
+        wants_introspection = bool(inspect.signature(controller.build).parameters)
+
+        if not wants_introspection:
+            await controller.build()
+            return
+
+        connection: Connection | None = controller.connection
+        if connection is None:
+            raise TypeError(
+                f"{type(controller).__name__}.build takes an introspection "
+                "argument, but the controller has no connection to get one from."
+            )
+
+        state = self._state.get(connection)
+        if state is None:
+            # A controller added during ``build`` that holds an unopened
+            # connection reaches here before the pass that would catch it, and a
+            # bare KeyError would say nothing useful.
+            raise self._unsupervised_connection_error(controller, connection)
+
+        await controller.build(state.introspection)  # type: ignore[call-arg]
+
+    def _check_connections_are_known(self) -> None:
+        """A connection the runner never opened would never be reconnected either."""
+        for controller in self._walk_controllers():
+            connection: Connection | None = controller.connection
+            if connection is None or connection in self._state:
                 continue
 
-            logger.info("Attempting to reconnect", controller=controller.path)
-            try:
-                await controller.reconnect()
-            except Exception:
-                logger.exception("Reconnect failed", controller=controller.path)
+            raise self._unsupervised_connection_error(controller, connection)
+
+    @staticmethod
+    def _unsupervised_connection_error(
+        controller: BaseController, connection: Connection
+    ) -> RuntimeError:
+        return RuntimeError(
+            f"Controller {'.'.join(controller.path) or type(controller).__name__} "
+            f"holds a {type(connection).__name__} the runner did not open. A "
+            "connection created during `build` cannot be supervised - declare it "
+            "up front and claim it from the `Connections` registry."
+        )
+
+    def _warn_about_unclaimed_connections(self) -> None:
+        if self._registry is None:
+            return
+
+        for name in sorted(self._registry.unclaimed()):
+            logger.warning(
+                "Connection declared but never used. It will be opened and "
+                "reconnected forever while doing nothing.",
+                connection=name,
+            )
+
+    def _warn_about_unpolled_connections(self) -> None:
+        """Nothing detects a connection failing unless something uses it regularly.
+
+        Phrased as fact rather than fault: an all-on-demand device is a legitimate
+        design, it just will not notice a failure until the next write.
+        """
+        polled: set[int] = set()
+        for controller in self._walk_controllers():
+            connection: Connection | None = controller.connection
+            if connection is None:
+                continue
+            if self._has_polling(controller):
+                polled.add(id(connection))
+
+        for connection in self._connections:
+            if id(connection) in polled:
+                continue
+
+            logger.warning(
+                "Connection has no polled attribute or scan method among its "
+                "controllers, so nothing will detect it failing until the next "
+                "write. It will not reconnect automatically.",
+                connection=self._name_of(connection),
+            )
+
+    @staticmethod
+    def _has_polling(controller: BaseController) -> bool:
+        from fastcs.attributes.attr_r import AttrR
+
+        for method in controller.scan_methods.values():
+            if method.period is not ONCE:
+                return True
+
+        for attribute in controller.attributes.values():
+            if not (isinstance(attribute, AttrR) and attribute.has_getter()):
+                continue
+            if attribute.poll_period is not ONCE and attribute.poll_period is not None:
+                return True
+
+        return False
+
+    # Failure and recovery
+
+    async def _reconnect_loop(self, connection: Connection) -> None:
+        """Keep one connection alive, at its own pace.
+
+        One task per connection, idle until that connection actually goes down - a
+        healthy connection costs nothing, and a detector that wants to retry every
+        five seconds does not have to compromise with a writer that wants one.
+        """
+        state = self._state[connection]
+
+        while True:
+            await connection.wait_down()
+
+            if state.exhausted.is_set():
+                return
+
+            # If what we ride on is down, wait for it rather than attempting. No
+            # attempt means no increment, so the retry budget freezes while waiting.
+            dependency = connection.depends_on
+            if dependency is not None and not dependency.connected:
+                logger.info(
+                    "Waiting on dependency",
+                    connection=self._name_of(connection),
+                    dependency=self._name_of(dependency),
+                )
+                await self._await_dependency(dependency)
+
+                if not dependency.connected:
+                    # The dependency gave up. This connection cannot succeed, but it
+                    # is not itself exhausted - it has spent nothing. Say so, then
+                    # wait; only a restart will change anything.
+                    logger.error(
+                        "Stalled: dependency gave up",
+                        connection=self._name_of(connection),
+                        dependency=self._name_of(dependency),
+                    )
+                    return
+
+            await self._attempt(connection)
+
+            if not connection.connected and not state.exhausted.is_set():
+                await asyncio.sleep(connection.reconnect_period)
+
+    async def _await_dependency(self, dependency: Connection) -> None:
+        """Block until the dependency either comes back or gives up.
+
+        Waiting on recovery alone would hang forever once the dependency exhausts, so
+        both outcomes are awaited and whichever lands first wins.
+        """
+        dependency_state = self._state[dependency]
+
+        recovered = asyncio.create_task(dependency.wait_up())
+        gave_up = asyncio.create_task(dependency_state.exhausted.wait())
+
+        _, pending = await asyncio.wait(
+            {recovered, gave_up}, return_when=asyncio.FIRST_COMPLETED
+        )
+        for task in pending:
+            task.cancel()
+
+    async def _attempt(self, connection: Connection) -> None:
+        """One reconnect attempt.
+
+        Owns retry accounting and the introspection check, and is the only place a
+        connection is marked back up.
+        """
+        state = self._state[connection]
+        state.attempts += 1
+
+        try:
+            await connection.close()  # tolerate an already-closed link
+            introspection = await connection.connect()
+        except Exception:
+            logger.exception("Reconnect failed", connection=self._name_of(connection))
+            if state.attempts >= connection.max_attempts:
+                # Terminal until the process restarts. Setting the event releases
+                # anything waiting on this connection, so dependents stall loudly
+                # instead of hanging silently.
+                state.exhausted.set()
+                logger.error(
+                    "Giving up",
+                    connection=self._name_of(connection),
+                    attempts=state.attempts,
+                    blocks=[
+                        self._name_of(dependent)
+                        for dependent in self._dependents_of(connection)
+                    ],
+                )
+            return
+
+        try:
+            differs = self._introspection_differs(introspection, state.introspection)
+        except TypeError as error:
+            # Raised for an introspection result that cannot be compared. Letting
+            # it out of here would kill this reconnect task silently - nothing
+            # awaits it - and every scan gated on this connection would then wait
+            # in `wait_up` forever. Report it the same way a mismatch is reported.
+            logger.exception(
+                "Cannot compare introspection", connection=self._name_of(connection)
+            )
+            self._fail(error)
+            return
+
+        if differs:
+            self._fatal_introspection_mismatch(
+                connection, state.introspection, introspection
+            )
+            return
+
+        connection._set_connected()  # noqa: SLF001
+        state.attempts = 0  # a clean connection restores the budget
+
+    @staticmethod
+    def _introspection_differs(new: object, old: object) -> bool:
+        """Whether a device is describing itself differently than it did at startup.
+
+        ``!=`` is the comparison, which means an introspection result has to compare
+        to a single bool - a dataclass does, an array of values does not. Saying so
+        beats an ``ambiguous truth value`` escaping from a background task.
+        """
+        try:
+            return bool(new != old)
+        except (TypeError, ValueError) as exc:
+            raise TypeError(
+                "Introspection results are compared with `!=` on every reconnect, "
+                f"but comparing {type(new).__name__} did not give a single bool. "
+                "Return something that compares by value, such as a dataclass of "
+                "plain fields."
+            ) from exc
+
+    def _fatal_introspection_mismatch(
+        self, connection: Connection, expected: object, received: object
+    ) -> None:
+        error = IntrospectionMismatchError(
+            f"Connection {self._name_of(connection)} came back describing itself "
+            f"differently: expected {expected!r}, got {received!r}. `build` cannot "
+            "run again, so the application cannot represent this device any more."
+        )
+        logger.error(
+            "Introspection mismatch on reconnect",
+            connection=self._name_of(connection),
+            expected=repr(expected),
+            received=repr(received),
+        )
+        self._fail(error)
+
+    def _fail(self, error: BaseException) -> None:
+        """Report a condition the runner cannot carry on from.
+
+        Raising here would be invisible - this runs in a background task with nothing
+        awaiting it - and an embedded FastCS must not call ``sys.exit``, so the
+        failure is recorded and whatever is running the runner decides what to do.
+        """
+        if self.fatal_reason is None:
+            self.fatal_reason = error
+        self.fatal_error.set()
+
+    def _dependents_of(self, connection: Connection) -> list[Connection]:
+        return [
+            other
+            for other in self._connections
+            if other.depends_on is connection  # identity: declared, not derived
+        ]
+
+    # Helpers
+
+    def _name_of(self, connection: Connection) -> str:
+        """What to call a connection in a log line."""
+        if self._registry is not None:
+            name = self._registry.name_of(connection)
+            if name is not None:
+                return name
+        return type(connection).__name__
+
+    def _walk_controllers(self) -> Iterator[BaseController]:
+        """Every controller in the tree, level order."""
+        queue: deque[BaseController] = deque(self._controllers)
+        while queue:
+            controller = queue.popleft()
+            yield controller
+            queue.extend(controller.sub_controllers.values())
 
     def _cancel_tasks(self) -> None:
         # ``Task.cancel`` does not raise - it returns whether the task was
diff --git a/src/fastcs/demo/eiger.py b/src/fastcs/demo/eiger.py
index ba89d1f13..8f44690ce 100644
--- a/src/fastcs/demo/eiger.py
+++ b/src/fastcs/demo/eiger.py
@@ -2,11 +2,15 @@
 
 Half the attributes (``count_time``, ``state``) are declared as type hints and
 checked by the current ``HintedAttribute`` introspection-validation mechanism; the
-rest of the parameter tree is discovered at ``initialise()`` time by walking the
-sim's ``keys`` endpoints and is added dynamically, with no static check. A device
-that describes itself over the wire is exactly the case where introspection earns
-its complexity - contrast with the (deliberately non-introspectable) SCPI/temperature
-examples.
+rest of the parameter tree is discovered by walking the sim's ``keys`` endpoints and
+is added dynamically, with no static check. A device that describes itself over the
+wire is exactly the case where introspection earns its complexity - contrast with the
+(deliberately non-introspectable) SCPI/temperature examples.
+
+The introspection happens in `EigerConnection.connect`, not in the controller, which
+is what earns the reconnect check: the connection returns a `DetectorInfo` that the
+framework keeps and compares on every reconnect, so a detector that comes back
+describing itself differently is caught rather than served stale.
 """
 
 import enum
@@ -16,6 +20,7 @@
 import httpx
 
 from fastcs.attributes import AttrR, AttrRW, Polled
+from fastcs.connections import Connection
 from fastcs.controllers import Controller
 from fastcs.datatypes import DType
 from fastcs.demo.simulation.eiger import API_PREFIX, Subsystem, ValueType
@@ -30,22 +35,51 @@
 # Poll period (seconds) for read-only status params that change on the device.
 UPDATE_PERIOD = 0.2
 
+SUBSYSTEMS: tuple[Subsystem, ...] = ("config", "status")
 
-def _datatype(param: str, data: dict[str, Any]) -> type[DType]:
+
+@dataclass(frozen=True)
+class ParameterInfo:
+    """What the device says about one of its parameters.
+
+    Deliberately the *shape* of the parameter and not its value: the value changes
+    every time it is read, and this is compared against the startup value on every
+    reconnect.
+    """
+
+    subsystem: Subsystem
+    name: str
+    value_type: ValueType
+    access_mode: str
+    allowed_values: tuple[str, ...] | None
+
+
+@dataclass(frozen=True)
+class DetectorInfo:
+    """Returned by `EigerConnection.connect`.
+
+    Compared against the startup value on every reconnect, so it must compare by
+    value - hence a frozen dataclass of plain fields rather than the raw JSON.
+    """
+
+    parameters: tuple[ParameterInfo, ...]
+
+
+def _datatype(info: ParameterInfo) -> type[DType]:
     """Build a datatype for a parameter from the metadata the device reports.
 
     A parameter that reports ``allowed_values`` is discrete, so it becomes an enum
     class built from those values. The members are only knowable over the wire,
     which is exactly the case introspection exists for.
     """
-    allowed_values = data.get("allowed_values")
-    if allowed_values is None:
-        return _DATATYPES[data["value_type"]]
+    if info.allowed_values is None:
+        return _DATATYPES[info.value_type]
 
-    name = "".join(part.title() for part in param.split("_"))
+    name = "".join(part.title() for part in info.name.split("_"))
     # The functional API builds a class; type checkers only see the instance signature.
     return cast(
-        type[enum.Enum], enum.Enum(name, {value: value for value in allowed_values})
+        type[enum.Enum],
+        enum.Enum(name, {value: value for value in info.allowed_values}),
     )
 
 
@@ -54,22 +88,59 @@ class EigerConnectionSettings:
     base_url: str = "http://localhost:8000"
 
 
-class EigerConnection:
-    """Thin async HTTP client wrapper for the Eiger REST sim.
+class EigerConnection(Connection[DetectorInfo]):
+    """HTTP to the Eiger REST sim, and the one thing that knows when it is down.
 
     A ``transport`` can be supplied to point directly at an in-process ASGI app
     (e.g. in tests), bypassing the network entirely.
+
+    Args:
+        settings: Where the detector's REST API lives
+        transport: Optional httpx transport, for talking to an in-process app
+        kwargs: Passed to `Connection`
+
     """
 
-    def __init__(self, transport: httpx.AsyncBaseTransport | None = None):
+    def __init__(
+        self,
+        settings: EigerConnectionSettings | None = None,
+        transport: httpx.AsyncBaseTransport | None = None,
+        **kwargs,
+    ) -> None:
+        super().__init__(**kwargs)
+        self._settings = settings or EigerConnectionSettings()
         self._transport = transport
         self._client: httpx.AsyncClient | None = None
 
-    async def connect(self, settings: EigerConnectionSettings) -> None:
+    async def connect(self) -> DetectorInfo:
+        """Open the client and ask the detector what it has.
+
+        Introspecting here rather than in a controller's ``build`` is what lets the
+        framework compare the answer on every reconnect.
+        """
         self._client = httpx.AsyncClient(
-            base_url=settings.base_url, transport=self._transport
+            base_url=self._settings.base_url, transport=self._transport
         )
 
+        parameters: list[ParameterInfo] = []
+        for subsystem in SUBSYSTEMS:
+            for param in await self.keys(subsystem):
+                data = await self.get(subsystem, param)
+                allowed_values = data.get("allowed_values")
+                parameters.append(
+                    ParameterInfo(
+                        subsystem=subsystem,
+                        name=param,
+                        value_type=data["value_type"],
+                        access_mode=data["access_mode"],
+                        allowed_values=(
+                            None if allowed_values is None else tuple(allowed_values)
+                        ),
+                    )
+                )
+
+        return DetectorInfo(parameters=tuple(parameters))
+
     async def close(self) -> None:
         if self._client is not None:
             await self._client.aclose()
@@ -81,30 +152,47 @@ def client(self) -> httpx.AsyncClient:
             raise RuntimeError("EigerConnection is not connected")
         return self._client
 
-    async def keys(self, subsystem: Subsystem) -> list[str]:
-        response = await self.client.get(f"{API_PREFIX}/{subsystem}/keys")
+    async def _request(self, method: str, url: str, **kwargs) -> httpx.Response:
+        """Every request goes through here, because this is where health is decided.
+
+        A transport error means the link is gone and everything holding this
+        connection is now down. A 4xx from the detector is a device complaint about
+        one parameter, and propagates to the caller without touching connection
+        state.
+        """
+        try:
+            response = await self.client.request(method, url, **kwargs)
+        except httpx.TransportError:
+            self.set_disconnected()
+            raise
+
         response.raise_for_status()
+        return response
+
+    async def keys(self, subsystem: Subsystem) -> list[str]:
+        response = await self._request("GET", f"{API_PREFIX}/{subsystem}/keys")
         return response.json()
 
     async def get(self, subsystem: Subsystem, param: str) -> dict:
-        response = await self.client.get(f"{API_PREFIX}/{subsystem}/{param}")
-        response.raise_for_status()
+        response = await self._request("GET", f"{API_PREFIX}/{subsystem}/{param}")
         return response.json()
 
     async def put(self, subsystem: Subsystem, param: str, value) -> None:
-        response = await self.client.put(
-            f"{API_PREFIX}/{subsystem}/{param}", json={"value": value}
+        await self._request(
+            "PUT", f"{API_PREFIX}/{subsystem}/{param}", json={"value": value}
         )
-        response.raise_for_status()
 
 
 class EigerDetector(Controller):
     """Cut-down Eiger controller: half declared, half introspected."""
 
+    connection: EigerConnection
+
     # Declared (checked): must exist, with this access mode and dtype, after
-    # initialise() introspects the parameter tree. ``state`` is discrete, and its
-    # enum class is built from the ``allowed_values`` the device reports, so there
-    # is no author-time type to hint - only the access mode can be pinned here.
+    # build() has turned the connection's introspection into attributes. ``state``
+    # is discrete, and its enum class is built from the ``allowed_values`` the
+    # device reports, so there is no author-time type to hint - only the access
+    # mode can be pinned here.
     count_time: AttrRW[float]
     state: AttrR
 
@@ -118,11 +206,9 @@ def __init__(
         settings: EigerConnectionSettings | None = None,
         transport: httpx.AsyncBaseTransport | None = None,
     ) -> None:
-        self.connection = EigerConnection(transport=transport)
+        self.connection = EigerConnection(settings, transport)
         super().__init__()
 
-        self._settings = settings or EigerConnectionSettings()
-
     def _getter(self, subsystem: Subsystem, param: str):
         async def get() -> Any:
             data = await self.connection.get(subsystem, param)
@@ -139,36 +225,30 @@ async def put(value: Any) -> None:
 
         return put
 
-    async def connect(self) -> None:
-        await self.connection.connect(self._settings)
-        self._connected = True
-
-    async def disconnect(self) -> None:
-        await self.connection.close()
-
-    async def initialise(self) -> None:
-        for subsystem in ("config", "status"):
-            for param in await self.connection.keys(subsystem):
-                data = await self.connection.get(subsystem, param)
-                datatype = _datatype(param, data)
-
-                if data["access_mode"] == "rw":
-                    attr = AttrRW(
-                        datatype,
-                        getter=self._getter(subsystem, param),
-                        setter=self._setter(subsystem, param),
-                    )
-                else:
-                    # Read-only params are status values that change on the device,
-                    # so poll them periodically rather than reading once.
-                    attr = AttrR(
-                        datatype,
-                        getter=Polled(
-                            self._getter(subsystem, param), period=UPDATE_PERIOD
-                        ),
-                    )
-
-                self.add_attribute(param, attr)
+    async def build(  # pyright: ignore[reportIncompatibleMethodOverride]
+        self, info: DetectorInfo
+    ) -> None:
+        """Turn what the connection found into attributes.
+
+        The argument is whatever ``EigerConnection.connect`` returned - a controller
+        that does not introspect writes ``build(self)`` instead.
+        """
+        for parameter in info.parameters:
+            datatype = _datatype(parameter)
+            getter = self._getter(parameter.subsystem, parameter.name)
+
+            if parameter.access_mode == "rw":
+                attr: AttrR = AttrRW(
+                    datatype,
+                    getter=getter,
+                    setter=self._setter(parameter.subsystem, parameter.name),
+                )
+            else:
+                # Read-only params are status values that change on the device,
+                # so poll them periodically rather than reading once.
+                attr = AttrR(datatype, getter=Polled(getter, period=UPDATE_PERIOD))
+
+            self.add_attribute(parameter.name, attr)
 
         # Keep the derived ``idle`` flag in sync with the introspected ``state``.
         self.state.add_readback_callback(self._update_idle)
diff --git a/src/fastcs/demo/temperature_attr.py b/src/fastcs/demo/temperature_attr.py
index 89a7caabc..4d1572bc2 100755
--- a/src/fastcs/demo/temperature_attr.py
+++ b/src/fastcs/demo/temperature_attr.py
@@ -126,8 +126,16 @@ async def get_actual(self) -> float:
 
 
 class TemperatureController(Controller):
+    # Narrows the base class's `Connection | None`, so this controller's own code
+    # can call the methods of the connection it actually holds.
+    connection: IPConnection
+
     def __init__(self, settings: TemperatureControllerSettings) -> None:
-        self.connection = IPConnection()
+        # The ramps below hold this same object rather than consulting this
+        # controller, so the whole tree has one health state and one reconnect task
+        # between it. Opening it, and reopening it after a failure, is the runner's
+        # job - nothing here connects.
+        self.connection = IPConnection(settings.ip_settings)
         self._settings = settings
         self._protocol = TemperatureProtocol(self.connection)
 
@@ -156,22 +164,6 @@ async def cancel_all(self) -> None:
             # TODO: The requests all get concatenated and the sim doesn't handle it
             await asyncio.sleep(0.1)
 
-    async def connect(self) -> None:
-        await self.connection.connect(self._settings.ip_settings)
-
-    async def reconnect(self):
-        try:
-            await self.connection.close()
-            await self.connection.connect(self._settings.ip_settings)
-        except BaseException:
-            logger.exception("Reconnect failed")
-            return
-
-        self._connected = True
-
-    async def close(self) -> None:
-        await self.connection.close()
-
     @scan(0.1)
     async def update_voltages(self):
         voltages = await self._protocol.get_voltages()
@@ -186,6 +178,8 @@ async def update_voltages(self):
 
 
 class TemperatureRampController(Controller):
+    connection: IPConnection
+
     def __init__(self, index: int, conn: IPConnection) -> None:
         self._protocol = TemperatureRampProtocol(conn, index)
 
diff --git a/tests/assertable_controller.py b/tests/assertable_controller.py
index 8299bff91..5c7993141 100644
--- a/tests/assertable_controller.py
+++ b/tests/assertable_controller.py
@@ -25,18 +25,12 @@ def __init__(self) -> None:
             self._sub_controllers.append(controller)
             self.add_sub_controller(f"SubController{index:02d}", controller)
 
-    initialised = False
+    built = False
     count = 0
 
-    async def initialise(self) -> None:
-        await super().initialise()
-        self.initialised = True
-
-    async def connect(self) -> None:
-        self._connected = True
-
-    async def disconnect(self) -> None:
-        self._connected = False
+    async def build(self) -> None:
+        await super().build()
+        self.built = True
 
     @command()
     async def go(self):
diff --git a/tests/demo/test_eiger.py b/tests/demo/test_eiger.py
index 2b518c8bb..095458cfa 100644
--- a/tests/demo/test_eiger.py
+++ b/tests/demo/test_eiger.py
@@ -6,6 +6,7 @@
 import pytest_asyncio
 
 from fastcs.attributes import AttrR, AttrRW
+from fastcs.controllers import ControllerRunner
 from fastcs.demo.eiger import UPDATE_PERIOD, EigerDetector
 from fastcs.demo.simulation.eiger import EigerParameter, create_eiger_sim_app
 from fastcs.util import ONCE
@@ -18,11 +19,10 @@
 async def _eiger():
     app = create_eiger_sim_app()
     controller = EigerDetector(transport=httpx.ASGITransport(app=app))
-    await controller.connect()
-    await controller.initialise()
-    controller.post_initialise()
+    runner = ControllerRunner(controller)
+    await runner.build()
     yield controller, app.state.sim
-    await controller.disconnect()
+    await runner.stop()
 
 
 @pytest_asyncio.fixture
@@ -120,9 +120,8 @@ async def test_temperature_oscillation_seen_via_subscribe():
     app = create_eiger_sim_app()
     async with app.router.lifespan_context(app):
         controller = EigerDetector(transport=httpx.ASGITransport(app=app))
-        await controller.connect()
-        await controller.initialise()
-        controller.post_initialise()
+        runner = ControllerRunner(controller)
+        await runner.build()
 
         temperature = controller.attributes["temperature"]
         assert isinstance(temperature, AttrR)
@@ -139,6 +138,6 @@ async def record(value: float) -> None:
             await temperature.poll()
             await asyncio.sleep(0.2)
 
-        await controller.disconnect()
+        await runner.stop()
 
     assert len(set(seen)) > 1, f"temperature did not change: {seen}"
diff --git a/tests/demo/test_temperature_attr.py b/tests/demo/test_temperature_attr.py
index bab260065..7e139bb2c 100644
--- a/tests/demo/test_temperature_attr.py
+++ b/tests/demo/test_temperature_attr.py
@@ -20,9 +20,7 @@ def controller() -> TemperatureController:
         num_ramp_controllers=4,
         ip_settings=IPConnectionSettings(ip="localhost", port=25565),
     )
-    controller = TemperatureController(settings)
-    controller.post_initialise()
-    return controller
+    return TemperatureController(settings)
 
 
 @pytest.fixture
diff --git a/tests/test_attributes.py b/tests/test_attributes.py
index 0a0b0d061..16a7db717 100644
--- a/tests/test_attributes.py
+++ b/tests/test_attributes.py
@@ -400,7 +400,7 @@ class DemoParameterController(Controller):
         int_parameter: AttrRW
         float_parameter: AttrRW  # hint to satisfy pyright
 
-        async def initialise(self):
+        async def build(self):
             self._connection = DummyConnection()
             await self._connection.connect()
             dtype_mapping = {"int": int, "float": float}
@@ -454,7 +454,7 @@ async def setter(value, uri=uri):
                     )
 
     c = DemoParameterController()
-    await c.initialise()
+    await c.build()
 
     assert await c.ro_int_parameter.poll() == 10
     assert await c.ro_int_parameter.poll() == 11
diff --git a/tests/test_connections.py b/tests/test_connections.py
new file mode 100644
index 000000000..e65d0e1ed
--- /dev/null
+++ b/tests/test_connections.py
@@ -0,0 +1,223 @@
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from fastcs.connections import (
+    Connection,
+    Connections,
+    IPConnection,
+    IPConnectionSettings,
+    SerialConnection,
+    SerialConnectionSettings,
+)
+from fastcs.connections.ip_connection import DisconnectedError, StreamConnection
+from fastcs.connections.serial_connection import NotOpenedError
+
+
+class OneConnection(Connection[None]):
+    async def connect(self) -> None: ...
+    async def close(self) -> None: ...
+
+
+class AnotherConnection(Connection[None]):
+    async def connect(self) -> None: ...
+    async def close(self) -> None: ...
+
+
+# Connections registry
+
+
+def test_a_connection_is_claimed_by_name_with_its_type_asserted():
+    connection = OneConnection()
+    registry = Connections({"one": connection})
+
+    assert registry.get("one", OneConnection) is connection
+
+
+def test_claiming_a_name_that_was_not_declared_lists_the_ones_that_were():
+    registry = Connections({"one": OneConnection(), "two": AnotherConnection()})
+
+    with pytest.raises(KeyError, match=r"No connection named 'three'") as exc:
+        registry.get("three", OneConnection)
+
+    assert "'one', 'two'" in str(exc.value)
+
+
+def test_claiming_a_name_with_the_wrong_type_says_both_types():
+    registry = Connections({"one": AnotherConnection()})
+
+    with pytest.raises(TypeError, match="is AnotherConnection, but OneConnection"):
+        registry.get("one", OneConnection)
+
+
+def test_a_registry_reports_what_was_never_claimed():
+    registry = Connections({"used": OneConnection(), "spare": OneConnection()})
+
+    assert registry.unclaimed() == {"used", "spare"}
+
+    registry.get("used", OneConnection)
+
+    assert registry.unclaimed() == {"spare"}
+
+
+def test_a_connection_is_named_by_identity_not_equality():
+    """Two connections with matching settings are two connections."""
+    first, second = OneConnection(), OneConnection()
+    registry = Connections({"first": first, "second": second})
+
+    assert registry.name_of(first) == "first"
+    assert registry.name_of(second) == "second"
+    assert registry.name_of(OneConnection()) is None
+
+
+def test_a_registry_keeps_declaration_order():
+    first, second = OneConnection(), AnotherConnection()
+    registry = Connections({"first": first, "second": second})
+
+    assert registry.values() == [first, second]
+    assert len(registry) == 2
+    assert "first" in registry
+    assert "third" not in registry
+    assert repr(registry) == "Connections(['first', 'second'])"
+
+
+# IPConnection
+
+
+@pytest.mark.asyncio
+async def test_ip_connect_opens_the_settings_it_was_given():
+    connection = IPConnection(IPConnectionSettings(ip="192.0.2.1", port=1234))
+    reader, writer = MagicMock(), MagicMock()
+
+    with patch(
+        "asyncio.open_connection", AsyncMock(return_value=(reader, writer))
+    ) as open_connection:
+        await connection.connect()
+
+    open_connection.assert_awaited_once_with("192.0.2.1", 1234)
+    assert isinstance(connection._connection, StreamConnection)  # noqa: SLF001
+
+
+@pytest.mark.asyncio
+async def test_using_an_unopened_ip_connection_says_so():
+    with pytest.raises(DisconnectedError, match="call connect"):
+        await IPConnection().send_command("ID?\r\n")
+
+
+@pytest.mark.asyncio
+async def test_a_command_that_hits_a_dead_socket_marks_the_link_down():
+    connection = IPConnection()
+    stream = MagicMock()
+    stream.__aenter__ = AsyncMock(return_value=stream)
+    stream.__aexit__ = AsyncMock(return_value=False)
+    stream.send_message = AsyncMock(side_effect=ConnectionResetError)
+    connection._IPConnection__connection = stream  # pyright: ignore[reportAttributeAccessIssue]
+    connection._set_connected()  # noqa: SLF001
+
+    with pytest.raises(ConnectionResetError):
+        await connection.send_command("R=1\r\n")
+
+    assert not connection.connected
+
+
+@pytest.mark.asyncio
+async def test_a_command_the_device_accepts_leaves_the_link_up():
+    connection = IPConnection()
+    stream = MagicMock()
+    stream.__aenter__ = AsyncMock(return_value=stream)
+    stream.__aexit__ = AsyncMock(return_value=False)
+    stream.send_message = AsyncMock()
+    connection._IPConnection__connection = stream  # pyright: ignore[reportAttributeAccessIssue]
+    connection._set_connected()  # noqa: SLF001
+
+    await connection.send_command("R=1\r\n")
+
+    stream.send_message.assert_awaited_once_with("R=1\r\n")
+    assert connection.connected
+
+
+@pytest.mark.asyncio
+async def test_stream_connection_reads_and_writes_lines():
+    reader = asyncio.StreamReader()
+    reader.feed_data(b"ID=1\r\n")
+    writer = MagicMock()
+    writer.drain = AsyncMock()
+    writer.wait_closed = AsyncMock()
+
+    stream = StreamConnection(reader, writer)
+    async with stream as held:
+        await held.send_message("ID?\r\n")
+        assert await held.receive_response() == "ID=1\r\n"
+
+    writer.write.assert_called_once_with(b"ID?\r\n")
+
+    await stream.close()
+    writer.close.assert_called_once()
+
+
+# SerialConnection
+
+
+@pytest.mark.asyncio
+async def test_serial_connect_opens_the_settings_it_was_given():
+    connection = SerialConnection(
+        SerialConnectionSettings(port="/dev/ttyS0", baud=9600)
+    )
+
+    with patch("aioserial.AioSerial") as aioserial:
+        await connection.connect()
+
+    aioserial.assert_called_once_with(port="/dev/ttyS0", baudrate=9600)
+
+
+@pytest.mark.asyncio
+async def test_using_an_unopened_serial_connection_says_so():
+    connection = SerialConnection(SerialConnectionSettings(port="/dev/ttyS0"))
+
+    with pytest.raises(NotOpenedError, match="call connect"):
+        await connection.send_command(b"ID?\r\n")
+
+
+@pytest.mark.asyncio
+async def test_serial_round_trip_leaves_the_link_up():
+    connection = SerialConnection(SerialConnectionSettings(port="/dev/ttyS0"))
+    stream = MagicMock()
+    stream.write_async = AsyncMock()
+    stream.read_async = AsyncMock(return_value=b"ID=1")
+
+    with patch("aioserial.AioSerial", return_value=stream):
+        await connection.connect()
+    connection._set_connected()  # noqa: SLF001
+
+    await connection.send_command(b"R=1\r\n")
+    assert await connection.send_query(b"ID?\r\n", 4) == b"ID=1"
+    assert connection.connected
+
+    await connection.close()
+    stream.close.assert_called_once()
+    # Closing an already-closed link is tolerated - the runner does it before
+    # every reconnect attempt.
+    await connection.close()
+
+
+@pytest.mark.asyncio
+async def test_a_serial_port_that_goes_away_marks_the_link_down():
+    connection = SerialConnection(SerialConnectionSettings(port="/dev/ttyS0"))
+    stream = MagicMock()
+    stream.write_async = AsyncMock(side_effect=OSError)
+    stream.read_async = AsyncMock(side_effect=OSError)
+
+    with patch("aioserial.AioSerial", return_value=stream):
+        await connection.connect()
+    connection._set_connected()  # noqa: SLF001
+
+    with pytest.raises(OSError):
+        await connection.send_command(b"R=1\r\n")
+    assert not connection.connected
+
+    connection._set_connected()  # noqa: SLF001
+    stream.write_async = AsyncMock()
+    with pytest.raises(OSError):
+        await connection.send_query(b"ID?\r\n", 4)
+    assert not connection.connected
diff --git a/tests/test_control_system.py b/tests/test_control_system.py
index 19e3215e7..f2cb1d618 100644
--- a/tests/test_control_system.py
+++ b/tests/test_control_system.py
@@ -3,8 +3,10 @@
 import pytest
 
 from fastcs.attributes import AttrR, NotPolled, Polled
+from fastcs.connections import Connection
 from fastcs.control_system import FastCS
 from fastcs.controllers import Controller
+from fastcs.controllers.runner import IntrospectionMismatchError
 from fastcs.methods import Command, command
 from fastcs.util import ONCE
 
@@ -30,7 +32,7 @@ class MyTestController(Controller):
         def __init__(self):
             super().__init__()
 
-        async def initialise(self):
+        async def build(self):
             async def do_nothing_dynamic() -> None:
                 pass
 
@@ -94,33 +96,81 @@ def __init__(self):
     assert controller.update_once.readback == 1
     assert controller.update_never.readback == 0
 
-    # One periodic scan task per distinct period, plus one reconnect watcher
+    # One periodic scan task per distinct period
     assert len(fastcs._runner._scan_coros) == 1
     assert len(fastcs._runner._initial_coros) == 1
 
 
 @pytest.mark.asyncio
-async def test_controller_connect_disconnect():
-    class MyTestController(Controller):
-        async def connect(self):
-            self.connect_called = True
+async def test_serve_opens_and_closes_the_connection():
+    """Opening and closing the link is the runner's job, not the controller's."""
 
-        async def disconnect(self):
-            self.connect_called = False
+    class MyTestConnection(Connection[None]):
+        def __init__(self):
+            super().__init__()
+            self.open = False
 
-    controller = MyTestController()
+        async def connect(self) -> None:
+            self.open = True
+
+        async def close(self) -> None:
+            self.open = False
+
+    class MyTestController(Controller):
+        def __init__(self, connection):
+            self.connection = connection
+            super().__init__()
+
+    connection = MyTestConnection()
+    controller = MyTestController(connection)
 
     loop = asyncio.get_event_loop()
     fastcs = FastCS(controller, [], loop)
 
     task = asyncio.create_task(fastcs.serve(interactive=False))
 
-    # connect is called at the start of serve
+    # The runner opens every connection at the start of serve
     await asyncio.sleep(0.1)
-    assert controller.connect_called
+    assert connection.open
+    assert controller.connected
 
     task.cancel()
 
-    # disconnect is called at the end of serve
+    # ...and closes them at the end of it
+    await asyncio.sleep(0.1)
+    assert not connection.open
+
+
+@pytest.mark.asyncio
+async def test_a_fatal_runner_condition_comes_out_of_serve():
+    """Not `sys.exit`: an embedded FastCS must be able to see this and decide."""
+
+    class MyTestConnection(Connection[str]):
+        def __init__(self):
+            super().__init__()
+            self.introspection = "v1"
+
+        async def connect(self) -> str:
+            return self.introspection
+
+        async def close(self) -> None: ...
+
+    class MyTestController(Controller):
+        def __init__(self, connection):
+            self.connection = connection
+            super().__init__()
+
+    connection = MyTestConnection()
+    connection.reconnect_period = 0.001
+    fastcs = FastCS(MyTestController(connection), [], asyncio.get_event_loop())
+
+    task = asyncio.create_task(fastcs.serve(interactive=False))
     await asyncio.sleep(0.1)
-    assert not controller.connect_called
+
+    # The device comes back describing itself differently, which `build` cannot
+    # be re-run to accommodate.
+    connection.introspection = "v2"
+    connection.set_disconnected()
+
+    with pytest.raises(IntrospectionMismatchError, match="describing itself"):
+        await asyncio.wait_for(task, timeout=5)
diff --git a/tests/test_controller_runner.py b/tests/test_controller_runner.py
index b60bb7a64..927fab5dd 100644
--- a/tests/test_controller_runner.py
+++ b/tests/test_controller_runner.py
@@ -3,33 +3,53 @@
 
 import pytest
 
-from fastcs.attributes import AttrR
+from fastcs.attributes import AttrR, Polled
+from fastcs.connections import (
+    DEFAULT_MAX_ATTEMPTS,
+    DEFAULT_RECONNECT_PERIOD,
+    Connection,
+    Connections,
+)
 from fastcs.controllers import Controller, ControllerRunner
-from fastcs.controllers.runner import RECONNECT_PERIOD
+from fastcs.controllers.runner import MAX_BUILD_PASSES, IntrospectionMismatchError
 from fastcs.methods import scan
 from fastcs.util import ONCE
 
 
+class FakeConnection(Connection[str]):
+    """A connection that opens when told to, and records what was asked of it."""
+
+    def __init__(self, introspection: str = "v1", **kwargs) -> None:
+        super().__init__(**kwargs)
+        self.introspection = introspection
+        self.fail_next: Exception | None = None
+        self.connects = 0
+        self.closes = 0
+
+    async def connect(self) -> str:
+        self.connects += 1
+        if self.fail_next is not None:
+            raise self.fail_next
+        return self.introspection
+
+    async def close(self) -> None:
+        self.closes += 1
+
+
 class LifecycleController(Controller):
     """Records every lifecycle hook the runner is supposed to call."""
 
-    def __init__(self):
+    def __init__(self, connection: Connection | None = None):
+        self.connection = connection
         super().__init__()
         self.events: list[str] = []
         self.count = AttrR(int)
 
-    async def initialise(self):
-        self.events.append("initialise")
-
-    def post_initialise(self):
-        self.events.append("post_initialise")
+    async def build(self):
+        self.events.append("build")
 
-    async def connect(self):
-        self.events.append("connect")
-        await super().connect()
-
-    async def disconnect(self):
-        self.events.append("disconnect")
+    async def setup(self):
+        self.events.append("setup")
 
     @scan(ONCE)
     async def read_once(self):
@@ -39,41 +59,60 @@ async def read_once(self):
 
 @pytest.mark.asyncio
 async def test_the_runner_drives_the_whole_lifecycle():
-    controller = LifecycleController()
+    connection = FakeConnection()
+    controller = LifecycleController(connection)
     runner = ControllerRunner(controller)
 
     await runner.start()
     try:
-        assert controller.events == [
-            "initialise",
-            "post_initialise",
-            "connect",
-            "initial",
-        ]
+        assert controller.events == ["build", "setup", "initial"]
         assert controller.count.readback == 1
+        assert connection.connects == 1
+        assert connection.connected
     finally:
         await runner.stop()
 
-    assert controller.events[-1] == "disconnect"
+    # Shutdown is a runner operation, not an author hook
+    assert connection.closes == 1
+
+
+@pytest.mark.asyncio
+async def test_connections_open_before_anything_is_built():
+    """A ``build`` runs against an open link, so it can ask the device questions."""
+    connection = FakeConnection()
+    order: list[str] = []
+
+    class RecordingController(Controller):
+        def __init__(self):
+            self.connection = connection
+            super().__init__()
+
+        async def build(self):
+            order.append(f"build(connected={connection.connected})")
+
+    runner = ControllerRunner(RecordingController())
+    await runner.build()
+
+    assert order == ["build(connected=True)"]
 
 
 @pytest.mark.asyncio
-async def test_setup_builds_the_apis_before_anything_connects():
-    """A transport is wired to the APIs between setup and start."""
-    controller = LifecycleController()
+async def test_build_builds_the_apis_before_anything_is_set_up():
+    """A transport is wired to the APIs between build and start."""
+    controller = LifecycleController(FakeConnection())
     runner = ControllerRunner(controller)
 
-    apis = await runner.setup()
+    apis = await runner.build()
 
     assert [api.path for api in apis] == [[]]
     assert "count" in apis[0].attributes
-    assert controller.events == ["initialise", "post_initialise"]
+    assert controller.events == ["build"]
     assert runner.controller_apis == apis
 
 
 @pytest.mark.asyncio
-async def test_start_sets_up_when_setup_has_not_run():
-    runner = ControllerRunner(LifecycleController())
+async def test_start_builds_when_build_has_not_run():
+    runner = ControllerRunner(LifecycleController(FakeConnection()))
 
     await runner.start()
     try:
@@ -84,7 +123,10 @@ async def test_start_sets_up_when_setup_has_not_run():
 
 @pytest.mark.asyncio
 async def test_a_runner_takes_several_controllers():
-    controllers = [LifecycleController(), LifecycleController()]
+    controllers = [
+        LifecycleController(FakeConnection()),
+        LifecycleController(FakeConnection()),
+    ]
     runner = ControllerRunner(controllers)
 
     await runner.start()
@@ -96,89 +138,439 @@ async def test_a_runner_takes_several_controllers():
 
 
 @pytest.mark.asyncio
-async def test_stop_reports_a_failing_disconnect_without_raising(monkeypatch):
-    class UndisconnectableController(LifecycleController):
-        async def disconnect(self):
-            raise RuntimeError("no")
+async def test_a_controller_with_no_connection_still_runs():
+    """A soft controller that groups others has nothing to connect."""
+    controller = LifecycleController(None)
+    runner = ControllerRunner(controller)
 
-    logged: list[tuple[str, BaseException | None]] = []
+    await runner.start()
+    try:
+        assert runner.connections == []
+        assert controller.events == ["build", "setup", "initial"]
+        assert controller.connected
+    finally:
+        await runner.stop()
 
-    def record_exception(event, **kwargs):
-        # ``logger.exception`` is called from the ``except`` block, so the
-        # exception it is reporting is the one currently being handled.
-        logged.append((event, sys.exc_info()[1]))
 
-    monkeypatch.setattr("fastcs.controllers.runner.logger.exception", record_exception)
+@pytest.mark.asyncio
+async def test_setup_runs_once_the_whole_tree_is_built():
+    """A parent's ``setup`` can read a child that only exists after ``build``."""
+    order: list[str] = []
 
-    runner = ControllerRunner(UndisconnectableController())
+    class Child(Controller):
+        async def build(self):
+            order.append("child build")
+
+        async def setup(self):
+            order.append("child setup")
+
+    class Parent(Controller):
+        async def build(self):
+            order.append("parent build")
+            self.add_sub_controller("CHILD", Child())
+
+        async def setup(self):
+            order.append("parent setup")
+
+    runner = ControllerRunner(Parent())
     await runner.start()
+    try:
+        assert order == ["parent build", "child build", "parent setup", "child setup"]
+    finally:
+        await runner.stop()
 
-    await runner.stop()
 
-    assert len(logged) == 1
-    event, error = logged[0]
-    assert event == "Exception during disconnect"
-    assert isinstance(error, RuntimeError)
-    assert str(error) == "no"
+@pytest.mark.asyncio
+async def test_build_repeats_until_the_tree_stops_growing():
+    class Tier(Controller):
+        def __init__(self, depth: int) -> None:
+            super().__init__()
+            self._depth = depth
+
+        async def build(self):
+            if self._depth:
+                self.add_sub_controller("SUB", Tier(self._depth - 1))
+
+    runner = ControllerRunner(Tier(3))
+    await runner.build()
+
+    controller = runner._controllers[0]
+    for _ in range(3):
+        controller = controller.sub_controllers["SUB"]  # type: ignore[assignment]
+    assert controller.sub_controllers == {}
 
 
 @pytest.mark.asyncio
-async def test_the_runner_reconnects_a_controller_that_dropped_out(monkeypatch):
-    """Nothing else calls reconnect, so a paused controller would stay paused."""
-    monkeypatch.setattr("fastcs.controllers.runner.RECONNECT_PERIOD", 0.01)
+async def test_a_tree_that_never_settles_is_caught():
+    class Runaway(Controller):
+        async def build(self):
+            self.add_sub_controller("SUB", Runaway())
 
-    class DroppingController(LifecycleController):
-        reconnects = 0
+    runner = ControllerRunner(Runaway())
 
-        async def reconnect(self):
-            self.reconnects += 1
-            await super().reconnect()
+    with pytest.raises(RuntimeError, match=f"{MAX_BUILD_PASSES} build passes"):
+        await runner.build()
 
-    controller = DroppingController()
-    runner = ControllerRunner(controller)
+
+@pytest.mark.asyncio
+async def test_build_receives_the_connections_introspection():
+    received: list[object] = []
+
+    class IntrospectingController(Controller):
+        def __init__(self):
+            self.connection = FakeConnection("api-1.8.0")
+            super().__init__()
+
+        async def build(self, info: str) -> None:  # pyright: ignore[reportIncompatibleMethodOverride]
+            received.append(info)
+
+    await ControllerRunner(IntrospectingController()).build()
+
+    assert received == ["api-1.8.0"]
+
+
+@pytest.mark.asyncio
+async def test_asking_for_introspection_without_a_connection_is_an_error():
+    class Confused(Controller):
+        async def build(self, info) -> None: ...  # pyright: ignore[reportIncompatibleMethodOverride]
+
+    with pytest.raises(TypeError, match="no connection"):
+        await ControllerRunner(Confused()).build()
+
+
+@pytest.mark.asyncio
+async def test_controllers_sharing_a_connection_are_one_connection():
+    """Identity, not equality: the tree is not the unit of failure, the link is."""
+    connection = FakeConnection()
+
+    class Parent(Controller):
+        def __init__(self):
+            self.connection = connection
+            super().__init__()
+            self.add_sub_controller("A", LifecycleController(connection))
+            self.add_sub_controller("B", LifecycleController(connection))
+
+    runner = ControllerRunner(Parent())
+    await runner.build()
+
+    assert runner.connections == [connection]
+    assert connection.connects == 1
+
+
+@pytest.mark.asyncio
+async def test_a_connection_created_during_build_is_rejected():
+    """It would never be opened, and so never reconnected either."""
+
+    class LateConnector(Controller):
+        async def build(self):
+            child = LifecycleController(FakeConnection())
+            self.add_sub_controller("LATE", child)
+
+    with pytest.raises(RuntimeError, match="did not open"):
+        await ControllerRunner(LateConnector()).build()
+
+
+@pytest.mark.asyncio
+async def test_a_declared_but_unclaimed_connection_is_warned_about(monkeypatch):
+    warnings: list[dict] = []
+    monkeypatch.setattr(
+        "fastcs.controllers.runner.logger.warning",
+        lambda event, **kwargs: warnings.append({"event": event, **kwargs}),
+    )
+
+    claimed = FakeConnection()
+    connections = Connections({"used": claimed, "spare": FakeConnection()})
+    connections.get("used", FakeConnection)
+
+    runner = ControllerRunner(LifecycleController(claimed), connections=connections)
     await runner.start()
     try:
-        assert controller.connected
+        assert [w["connection"] for w in warnings if "never used" in w["event"]] == [
+            "spare"
+        ]
+    finally:
+        await runner.stop()
+
+
+@pytest.mark.asyncio
+async def test_a_connection_nothing_polls_is_warned_about(monkeypatch):
+    warnings: list[dict] = []
+    monkeypatch.setattr(
+        "fastcs.controllers.runner.logger.warning",
+        lambda event, **kwargs: warnings.append({"event": event, **kwargs}),
+    )
+
+    class OnDemandOnly(Controller):
+        def __init__(self, connection):
+            self.connection = connection
+            super().__init__()
+
+    runner = ControllerRunner(OnDemandOnly(FakeConnection()))
+    await runner.start()
+    try:
+        assert any("no polled attribute" in w["event"] for w in warnings)
+    finally:
+        await runner.stop()
+
+    warnings.clear()
+
+    class Polling(Controller):
+        def __init__(self, connection):
+            self.connection = connection
+            super().__init__()
+            self.value = AttrR(int, getter=Polled(self._get, period=0.2))
+
+        async def _get(self) -> int:
+            return 1
+
+    runner = ControllerRunner(Polling(FakeConnection()))
+    await runner.start()
+    try:
+        assert not any("no polled attribute" in w["event"] for w in warnings)
+    finally:
+        await runner.stop()
+
+
+@pytest.mark.asyncio
+async def test_the_runner_reconnects_a_connection_that_dropped_out():
+    """The connection's own IO marks it down; one task per connection brings it back."""
+    connection = FakeConnection(reconnect_period=0.01)
+    runner = ControllerRunner(LifecycleController(connection))
+    await runner.start()
+    try:
+        assert connection.connected
+
+        # What a connection's IO does when its transport fails
+        connection.set_disconnected()
+        await connection.wait_up()
+
+        assert connection.connected
+        assert connection.connects == 2
+        # Closed before the reconnect attempt, tolerating an already-closed link
+        assert connection.closes == 1
+    finally:
+        await runner.stop()
+
 
-        # What a scan task does when its callback raises
-        controller._connected = False
+@pytest.mark.asyncio
+async def test_a_failing_reconnect_keeps_trying_then_gives_up():
+    connection = FakeConnection(reconnect_period=0.001, max_attempts=3)
+    runner = ControllerRunner(LifecycleController(connection))
+    await runner.start()
+    try:
+        connection.fail_next = RuntimeError("still down")
+        connection.set_disconnected()
+
+        state = runner._state[connection]
+        await asyncio.wait_for(state.exhausted.wait(), timeout=2)
 
+        assert state.attempts == 3
+        assert not connection.connected
+
+        # Terminal until the process restarts: no further attempts
+        attempts_at_exhaustion = connection.connects
         await asyncio.sleep(0.05)
+        assert connection.connects == attempts_at_exhaustion
+    finally:
+        await runner.stop()
 
-        assert controller.reconnects >= 1
-        assert controller.connected
+
+@pytest.mark.asyncio
+async def test_a_clean_reconnect_restores_the_retry_budget():
+    connection = FakeConnection(reconnect_period=0.001, max_attempts=1000)
+    runner = ControllerRunner(LifecycleController(connection))
+    await runner.start()
+    try:
+        connection.fail_next = RuntimeError("down")
+        connection.set_disconnected()
+        await asyncio.sleep(0.02)
+        assert runner._state[connection].attempts > 0
+
+        connection.fail_next = None
+        await asyncio.wait_for(connection.wait_up(), timeout=2)
+
+        assert runner._state[connection].attempts == 0
     finally:
         await runner.stop()
 
 
 @pytest.mark.asyncio
-async def test_a_failing_reconnect_does_not_stop_the_runner(monkeypatch):
-    monkeypatch.setattr("fastcs.controllers.runner.RECONNECT_PERIOD", 0.01)
+async def test_a_dependent_waits_rather_than_spending_its_budget():
+    """No attempt means no increment, so the budget freezes while waiting."""
+    base = FakeConnection(reconnect_period=0.001, max_attempts=1000)
+    layered = FakeConnection(depends_on=base, reconnect_period=0.001)
 
-    class UnreconnectableController(LifecycleController):
-        attempts = 0
+    runner = ControllerRunner([LifecycleController(base), LifecycleController(layered)])
+    await runner.start()
+    try:
+        base.fail_next = RuntimeError("down")
+        base.set_disconnected()
+        layered.set_disconnected()
+
+        await asyncio.sleep(0.05)
+
+        # The dependent has not attempted at all while its dependency is down
+        assert runner._state[layered].attempts == 0
+        assert layered.connects == 1
+
+        base.fail_next = None
+        await asyncio.wait_for(layered.wait_up(), timeout=2)
+    finally:
+        await runner.stop()
 
-        async def reconnect(self):
-            self.attempts += 1
-            raise RuntimeError("still down")
 
-    controller = UnreconnectableController()
+@pytest.mark.asyncio
+async def test_a_dependent_is_released_when_its_dependency_gives_up(monkeypatch):
+    """Released rather than left hanging, so it stalls loudly."""
+    errors: list[dict] = []
+    monkeypatch.setattr(
+        "fastcs.controllers.runner.logger.error",
+        lambda event, **kwargs: errors.append({"event": event, **kwargs}),
+    )
+
+    base = FakeConnection(reconnect_period=0.001, max_attempts=1)
+    layered = FakeConnection(depends_on=base, reconnect_period=0.001)
+
+    runner = ControllerRunner(
+        [LifecycleController(base), LifecycleController(layered)],
+        connections=Connections({"base": base, "layered": layered}),
+    )
+    await runner.start()
+    try:
+        base.fail_next = RuntimeError("down for good")
+        base.set_disconnected()
+        layered.set_disconnected()
+
+        await asyncio.sleep(0.2)
+
+        gave_up = [e for e in errors if e["event"] == "Giving up"]
+        assert gave_up and gave_up[0]["connection"] == "base"
+        # The give-up message names what it takes down with it
+        assert gave_up[0]["blocks"] == ["layered"]
+
+        stalled = [e for e in errors if e["event"].startswith("Stalled")]
+        assert stalled and stalled[0]["connection"] == "layered"
+    finally:
+        await runner.stop()
+
+
+@pytest.mark.asyncio
+async def test_a_dependency_cycle_is_caught_at_startup():
+    first = FakeConnection()
+    second = FakeConnection(depends_on=first)
+    first.depends_on = second
+
+    runner = ControllerRunner([LifecycleController(first), LifecycleController(second)])
+
+    with pytest.raises(ValueError, match="Cycle in connection dependencies"):
+        await runner.build()
+
+
+@pytest.mark.asyncio
+async def test_a_device_that_comes_back_different_is_fatal():
+    connection = FakeConnection("v1", reconnect_period=0.001)
+    runner = ControllerRunner(LifecycleController(connection))
+    await runner.start()
+    try:
+        connection.introspection = "v2"
+        connection.set_disconnected()
+
+        await asyncio.wait_for(runner.fatal_error.wait(), timeout=2)
+
+        assert isinstance(runner.fatal_reason, IntrospectionMismatchError)
+        assert "v1" in str(runner.fatal_reason)
+        assert "v2" in str(runner.fatal_reason)
+        # Not marked back up against a tree that no longer matches the hardware
+        assert not connection.connected
+    finally:
+        await runner.stop()
+
+
+@pytest.mark.asyncio
+async def test_scans_are_gated_on_the_connection():
+    connection = FakeConnection(reconnect_period=0.001)
+
+    class Scanning(Controller):
+        def __init__(self):
+            self.connection = connection
+            super().__init__()
+            self.scans = 0
+
+        @scan(0.001)
+        async def tick(self):
+            self.scans += 1
+
+    controller = Scanning()
     runner = ControllerRunner(controller)
     await runner.start()
     try:
-        controller._connected = False
-        await asyncio.sleep(0.05)
+        await asyncio.sleep(0.02)
+        assert controller.scans > 0
+
+        connection.fail_next = RuntimeError("down")
+        connection.set_disconnected()
+        await asyncio.sleep(0.02)
 
-        # It keeps trying rather than dying on the first failure
-        assert controller.attempts > 1
-        assert not controller.connected
+        paused_at = controller.scans
+        await asyncio.sleep(0.02)
+        assert controller.scans == paused_at
     finally:
         await runner.stop()
 
 
+@pytest.mark.asyncio
+async def test_stop_closes_connections_in_reverse_declaration_order():
+    closed: list[str] = []
+
+    class Recording(FakeConnection):
+        def __init__(self, name: str, **kwargs):
+            super().__init__(**kwargs)
+            self.name = name
+
+        async def close(self) -> None:
+            await super().close()
+            closed.append(self.name)
+
+    base = Recording("base")
+    layered = Recording("layered", depends_on=base)
+
+    runner = ControllerRunner([LifecycleController(base), LifecycleController(layered)])
+    await runner.start()
+    await runner.stop()
+
+    assert closed == ["layered", "base"]
+
+
+@pytest.mark.asyncio
+async def test_stop_reports_a_failing_close_without_raising(monkeypatch):
+    class UncloseableConnection(FakeConnection):
+        async def close(self) -> None:
+            raise RuntimeError("no")
+
+    logged: list[tuple[str, BaseException | None]] = []
+
+    def record_exception(event, **kwargs):
+        # ``logger.exception`` is called from the ``except`` block, so the
+        # exception it is reporting is the one currently being handled.
+        logged.append((event, sys.exc_info()[1]))
+
+    monkeypatch.setattr("fastcs.controllers.runner.logger.exception", record_exception)
+
+    runner = ControllerRunner(LifecycleController(UncloseableConnection()))
+    await runner.start()
+
+    await runner.stop()
+
+    assert len(logged) == 1
+    event, error = logged[0]
+    assert event == "Exception while closing connection"
+    assert isinstance(error, RuntimeError)
+    assert str(error) == "no"
+
+
 @pytest.mark.asyncio
 async def test_stop_cancels_the_tasks():
-    controller = LifecycleController()
+    controller = LifecycleController(FakeConnection())
     runner = ControllerRunner(controller)
     await runner.start()
     tasks = set(runner._tasks)
@@ -191,5 +583,116 @@ async def test_stop_cancels_the_tasks():
     assert not runner._tasks
 
 
-def test_reconnect_period_is_a_second_by_default():
-    assert RECONNECT_PERIOD == 1.0
+def test_the_framework_connection_defaults():
+    assert DEFAULT_RECONNECT_PERIOD == 1.0
+    assert DEFAULT_MAX_ATTEMPTS == 10
+
+    connection = FakeConnection()
+    assert connection.reconnect_period == DEFAULT_RECONNECT_PERIOD
+    assert connection.max_attempts == DEFAULT_MAX_ATTEMPTS
+
+
+def test_a_class_default_sits_between_the_framework_and_the_constructor():
+    class Patient(FakeConnection):
+        reconnect_period = 5.0
+        max_attempts = 60
+
+    assert Patient().reconnect_period == 5.0
+    assert Patient().max_attempts == 60
+    assert Patient(reconnect_period=0.5).reconnect_period == 0.5
+    assert Patient(max_attempts=2).max_attempts == 2
+
+
+@pytest.mark.asyncio
+async def test_a_failed_build_closes_what_it_opened():
+    """Startup aborts, and no task exists yet for a later `stop` to clean up."""
+    connection = FakeConnection()
+
+    class Unbuildable(Controller):
+        def __init__(self):
+            self.connection = connection
+            super().__init__()
+
+        async def build(self):
+            raise RuntimeError("cannot build")
+
+    runner = ControllerRunner(Unbuildable())
+
+    with pytest.raises(RuntimeError, match="cannot build"):
+        await runner.build()
+
+    assert connection.closes == 1
+
+
+@pytest.mark.asyncio
+async def test_a_failed_setup_closes_what_it_opened():
+    connection = FakeConnection()
+
+    class Unsetuppable(Controller):
+        def __init__(self):
+            self.connection = connection
+            super().__init__()
+
+        async def setup(self):
+            raise RuntimeError("cannot set up")
+
+    runner = ControllerRunner(Unsetuppable())
+
+    with pytest.raises(RuntimeError, match="cannot set up"):
+        await runner.start()
+
+    assert connection.closes == 1
+
+
+@pytest.mark.asyncio
+async def test_a_dependency_the_runner_does_not_supervise_is_rejected():
+    """It would never be opened, so the dependent could never be attempted."""
+    unsupervised = FakeConnection()
+    layered = FakeConnection(depends_on=unsupervised)
+
+    runner = ControllerRunner(LifecycleController(layered))
+
+    with pytest.raises(ValueError, match="does not supervise"):
+        await runner.build()
+
+
+@pytest.mark.asyncio
+async def test_an_unopened_connection_in_call_build_says_what_is_wrong():
+    """A bare KeyError here would hide the diagnostic written for this case."""
+
+    class LateIntrospector(Controller):
+        def __init__(self):
+            self.connection = FakeConnection()
+            super().__init__()
+
+        async def build(self, info: str) -> None:  # pyright: ignore[reportIncompatibleMethodOverride]
+            ...
+
+    class Parent(Controller):
+        async def build(self):
+            self.add_sub_controller("LATE", LateIntrospector())
+
+    with pytest.raises(RuntimeError, match="did not open"):
+        await ControllerRunner(Parent()).build()
+
+
+@pytest.mark.asyncio
+async def test_an_uncomparable_introspection_result_is_reported_not_raised():
+    """Raising here would kill the reconnect task silently, hanging every scan."""
+
+    class Ambiguous:
+        def __ne__(self, other):
+            raise ValueError("truth value of an array is ambiguous")
+
+    connection = FakeConnection(Ambiguous(), reconnect_period=0.001)  # type: ignore[arg-type]
+    runner = ControllerRunner(LifecycleController(connection))
+    await runner.start()
+    try:
+        connection.set_disconnected()
+
+        await asyncio.wait_for(runner.fatal_error.wait(), timeout=2)
+
+        assert isinstance(runner.fatal_reason, TypeError)
+        assert "did not give a single bool" in str(runner.fatal_reason)
+    finally:
+        await runner.stop()
diff --git a/tests/test_controllers.py b/tests/test_controllers.py
index c6d2a9ee9..d25a0d490 100644
--- a/tests/test_controllers.py
+++ b/tests/test_controllers.py
@@ -4,6 +4,7 @@
 import pytest
 
 from fastcs.attributes import AttrR, AttrRW
+from fastcs.connections import Connection
 from fastcs.controllers import Controller, ControllerVector
 from fastcs.methods import Command, Scan, command, scan
 
@@ -257,26 +258,69 @@ async def scan_nothing(self):
 
 
 @pytest.mark.asyncio
-async def test_scan_exception_sets_disconnected_and_reconnect_resumes():
+async def test_a_raising_scan_is_logged_and_retried():
+    """A scan does not decide the connection is down - the connection's IO does."""
+    calls = 0
+
     class MyTestController(Controller):
         @scan(0.01)
         async def failing_scan(self):
+            nonlocal calls
+            calls += 1
             raise RuntimeError("scan error")
 
     controller = MyTestController()
-    controller.post_initialise()
     _, scan_coros, _ = controller.create_api_and_tasks()
 
-    controller._connected = True
     task = asyncio.create_task(scan_coros[0]())
-
-    # Wait long enough for the scan to run and raise, setting _connected = False
     await asyncio.sleep(0.1)
-    assert not controller._connected
 
-    # Trigger reconnect - _connected resumes scan tasks
-    await controller.reconnect()
-    assert controller._connected
+    assert calls > 1
+    assert controller.connected  # no connection to be down
+
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+
+
+@pytest.mark.asyncio
+async def test_scans_wait_while_the_connection_is_down():
+    class MyTestConnection(Connection[None]):
+        async def connect(self) -> None: ...
+        async def close(self) -> None: ...
+
+    connection = MyTestConnection()
+    calls = 0
+
+    class MyTestController(Controller):
+        def __init__(self):
+            self.connection = connection
+            super().__init__()
+
+        @scan(0.01)
+        async def counting_scan(self):
+            nonlocal calls
+            calls += 1
+
+    controller = MyTestController()
+    _, scan_coros, _ = controller.create_api_and_tasks()
+
+    task = asyncio.create_task(scan_coros[0]())
+
+    # Starts life down, so nothing runs until the framework marks it up
+    await asyncio.sleep(0.05)
+    assert calls == 0
+    assert not controller.connected
+
+    connection._set_connected()
+    await asyncio.sleep(0.05)
+    assert calls > 0
+
+    connection.set_disconnected()
+    await asyncio.sleep(0.02)
+    paused_at = calls
+    await asyncio.sleep(0.05)
+    assert calls == paused_at
 
     task.cancel()
     with pytest.raises(asyncio.CancelledError):
diff --git a/tests/test_ip_connection.py b/tests/test_ip_connection.py
index 3770174f1..3a50eac2a 100644
--- a/tests/test_ip_connection.py
+++ b/tests/test_ip_connection.py
@@ -2,7 +2,7 @@
 
 import pytest
 
-from fastcs.connections.ip_connection import IPConnection
+from fastcs.connections.ip_connection import DisconnectedError, IPConnection
 
 
 @pytest.fixture
@@ -50,3 +50,38 @@ async def test_close_connected_and_connection_reset(connection):
         await conn.close()
 
     assert conn._IPConnection__connection is None
+
+
+@pytest.mark.asyncio
+async def test_a_peer_that_closes_instead_of_answering_marks_the_link_down():
+    """``readline`` returns b"" at EOF, which is a dead link, not an empty reply."""
+    conn = IPConnection()
+    mock_stream = MagicMock()
+    mock_stream.__aenter__ = AsyncMock(return_value=mock_stream)
+    mock_stream.__aexit__ = AsyncMock(return_value=False)
+    mock_stream.send_message = AsyncMock()
+    mock_stream.receive_response = AsyncMock(return_value="")
+    conn._IPConnection__connection = mock_stream  # pyright: ignore[reportAttributeAccessIssue]
+    conn._set_connected()
+
+    with pytest.raises(DisconnectedError):
+        await conn.send_query("ID?\r\n")
+
+    # Without this the caller just gets "", fails to parse it, and retries forever
+    # while the reconnect task stays idle.
+    assert not conn.connected
+
+
+@pytest.mark.asyncio
+async def test_a_real_response_leaves_the_link_up():
+    conn = IPConnection()
+    mock_stream = MagicMock()
+    mock_stream.__aenter__ = AsyncMock(return_value=mock_stream)
+    mock_stream.__aexit__ = AsyncMock(return_value=False)
+    mock_stream.send_message = AsyncMock()
+    mock_stream.receive_response = AsyncMock(return_value="ID=1\r\n")
+    conn._IPConnection__connection = mock_stream  # pyright: ignore[reportAttributeAccessIssue]
+    conn._set_connected()
+
+    assert await conn.send_query("ID?\r\n") == "ID=1\r\n"
+    assert conn.connected
diff --git a/tests/test_multi_controller.py b/tests/test_multi_controller.py
index 1edf30f72..644dc4690 100644
--- a/tests/test_multi_controller.py
+++ b/tests/test_multi_controller.py
@@ -10,6 +10,7 @@
 from pytest_mock import MockerFixture
 
 from fastcs.attributes import AttrR
+from fastcs.connections import Connection
 from fastcs.control_system import FastCS
 from fastcs.controllers import Controller
 from fastcs.transports.epics import EpicsDocsOptions, EpicsGUIOptions
@@ -299,28 +300,38 @@ class names, so ``DEV-1`` and ``DEV_1`` would silently override each other in
     assert "'DEV_1'" in message
 
 
+class _LifecycleConnection(Connection[None]):
+    """Records whether the runner opened and closed the link."""
+
+    def __init__(self):
+        super().__init__()
+        self.open = False
+
+    async def connect(self) -> None:
+        self.open = True
+
+    async def close(self) -> None:
+        self.open = False
+
+
 class _LifecycleController(Controller):
     """Records lifecycle hook calls for end-to-end assertions."""
 
+    connection: _LifecycleConnection
+
     foo = AttrR(int)
 
     def __init__(self):
+        self.connection = _LifecycleConnection()
         super().__init__()
-        self.connect_called = False
-        self.initialised = False
-        self.post_initialised = False
-
-    async def initialise(self):
-        self.initialised = True
-
-    def post_initialise(self):
-        self.post_initialised = True
+        self.built = False
+        self.set_up = False
 
-    async def connect(self):
-        self.connect_called = True
+    async def build(self):
+        self.built = True
 
-    async def disconnect(self):
-        self.connect_called = False
+    async def setup(self):
+        self.set_up = True
 
 
 class _OtherLifecycleController(_LifecycleController):
@@ -347,9 +358,9 @@ async def test_fastcs_serves_two_controllers_end_to_end(mocker: MockerFixture):
         await asyncio.sleep(0.1)
 
         for controller in (a, b):
-            assert controller.initialised
-            assert controller.post_initialised
-            assert controller.connect_called
+            assert controller.built
+            assert controller.set_up
+            assert controller.connection.open
 
         with TestClient(transport._server._app) as client:
             assert client.get("/alpha/foo").status_code == 200
@@ -369,4 +380,4 @@ async def test_fastcs_serves_two_controllers_end_to_end(mocker: MockerFixture):
             pass
 
     for controller in (a, b):
-        assert not controller.connect_called
+        assert not controller.connection.open
diff --git a/tests/transports/epics/pva/test_p4p.py b/tests/transports/epics/pva/test_p4p.py
index 56fe9b897..0206e9cd9 100644
--- a/tests/transports/epics/pva/test_p4p.py
+++ b/tests/transports/epics/pva/test_p4p.py
@@ -700,8 +700,7 @@ async def get_a(self) -> int:
 
     controller = SeedController()
     controller.set_path([str(uuid4())])
-    await controller.initialise()
-    controller.post_initialise()
+    await controller.build()
     controller_api, _, initial_coros = controller.create_api_and_tasks()
 
     attribute = controller_api.attributes["a"]