diff --git a/docs/contents/specification/connector.md b/docs/contents/specification/connector.md
index 6c9140d..1975686 100644
--- a/docs/contents/specification/connector.md
+++ b/docs/contents/specification/connector.md
@@ -31,7 +31,7 @@ Typical responsibilities:
 
 - Connect to your fleet manager / backend services.
 - Optionally fetch the fleet membership and call `update_fleet()` **before** sessions are created.
-- Start background polling tasks that keep fresh state for all robots.
+- Start background polling tasks that keep fresh state for all robots. Schedule them with [`_create_supervised_task()`](#spec-connector-fleetconnector-background-tasks) rather than a bare `asyncio.create_task`, so a crash is logged and the loop restarts instead of dying silently.
 
 <a id="spec-connector-fleetconnector-execution-loop"></a>
 ### `_execution_loop()`
@@ -50,7 +50,19 @@ Typical responsibilities:
 
 **Override.**
 
-Called once during shutdown, after Edge SDK sessions are disconnected. Use this to stop polling tasks, close sockets, and release resources.
+Called once during shutdown, after Edge SDK sessions are disconnected. Use this to stop polling tasks, close sockets, and release resources. (Tasks scheduled via [`_create_supervised_task()` / `_spawn_logged_task()`](#spec-connector-fleetconnector-background-tasks) are cancelled automatically — you only need to clean up tasks you started with a bare `asyncio.create_task`, which is discouraged.)
+
+<a id="spec-connector-fleetconnector-background-tasks"></a>
+### `_create_supervised_task(name, coro_factory, restart_delay=5.0)` / `_spawn_logged_task(name, coro)`
+
+**Callable (advanced).**
+
+Schedule background work that runs alongside `_execution_loop` — for example a fast pose loop and a slower key-value loop, each on its own cadence. Prefer these over a bare `asyncio.create_task`: a fire-and-forget task that raises has its exception stored on a task nobody awaits, so it is never logged, never restarted, and the datasource it fed silently freezes until the process restarts (while unrelated datasources keep working, masking the failure).
+
+- **`_create_supervised_task(name, coro_factory, restart_delay=5.0)`** — for long-lived loops. Runs `coro_factory()` and, if it ever returns or raises, logs it (with a traceback, tagged with `name`) and restarts it after `restart_delay` seconds. `coro_factory` is a zero-argument callable returning a fresh coroutine (it is re-called on each restart). `asyncio.CancelledError` is propagated so the task stops cleanly on shutdown.
+- **`_spawn_logged_task(name, coro)`** — for one-shot work. Runs `coro` once; if it raises, the failure is logged (with a traceback, tagged with `name`) via a done-callback instead of being silently swallowed. It is **not** restarted.
+
+Call these from `_connect()` (they require the connector's event loop to be running). Tasks scheduled this way are tracked by the framework and cancelled automatically during shutdown, so they do not need to be stopped in `_disconnect()`.
 
 <a id="spec-connector-fleetconnector-command-handler"></a>
 ### `_inorbit_robot_command_handler(robot_id, command_name, args, options)`
diff --git a/docs/contents/usage/fleet.md b/docs/contents/usage/fleet.md
index 41cb0e3..4e0113e 100644
--- a/docs/contents/usage/fleet.md
+++ b/docs/contents/usage/fleet.md
@@ -114,6 +114,30 @@ All publishing methods require a `robot_id` parameter. See the [Publishing Guide
 - `publish_robot_system_stats(robot_id, **kwargs)`: Defer publishing of system stats for a specific robot; defaults are published if not called
 - `publish_robot_map(robot_id, frame_id, is_update=False)`: Publish map for a specific robot
 
+## Background tasks
+
+For work that needs a different cadence than `_execution_loop` (e.g. a fast per-robot poll), schedule **supervised** background loops from `_connect()` instead of bare `asyncio.create_task` calls. The framework logs and restarts a loop if it crashes — instead of the task dying silently and freezing that data source — and cancels it automatically on shutdown:
+
+```python
+@override
+async def _connect(self) -> None:
+    self._fleet_api = MyFleetApi(...)
+    # One supervised loop per robot. coro_factory is zero-arg, so bind the
+    # robot id with a default-arg lambda:
+    for robot_id in self.robot_ids:
+        self._create_supervised_task(
+            f"poll:{robot_id}", lambda rid=robot_id: self._poll_robot(rid)
+        )
+
+async def _poll_robot(self, robot_id: str) -> None:
+    while True:
+        speed = await self._fleet_api.get_speed(robot_id)
+        self.publish_robot_odometry(robot_id, linear_speed=speed)
+        await asyncio.sleep(0.2)
+```
+
+For one-shot work, use `_spawn_logged_task(name, coro)` — its failure is logged (not silently swallowed), but it is not restarted. See the [Connector API spec](../specification/connector#spec-connector-fleetconnector-background-tasks).
+
 ## Advanced Methods
 
 ### `_get_robot_session()`
diff --git a/docs/contents/usage/single-robot.md b/docs/contents/usage/single-robot.md
index f028b6c..e6cceb7 100644
--- a/docs/contents/usage/single-robot.md
+++ b/docs/contents/usage/single-robot.md
@@ -61,6 +61,28 @@ async def _execution_loop(self) -> None:
 
 The loop runs at the frequency specified by `config.update_freq` (default: 1.0 Hz).
 
+### Background tasks
+
+For work that needs a different cadence than `_execution_loop` (e.g. a fast sensor poll), schedule a **supervised** background loop from `_connect()` instead of a bare `asyncio.create_task`. The framework logs and restarts it if it crashes — instead of the task dying silently and freezing that data source — and cancels it automatically on shutdown:
+
+```python
+@override
+async def _connect(self) -> None:
+    self._robot_api = MyRobotApi(...)
+    # Long-lived loop on its own cadence, supervised (logged + restarted on crash):
+    self._create_supervised_task("speed_poll", self._poll_speeds)
+
+async def _poll_speeds(self) -> None:
+    while True:
+        speed = await self._robot_api.get_speed()
+        self.publish_odometry(linear_speed=speed)
+        await asyncio.sleep(0.2)
+```
+
+For one-shot startup work, use `_spawn_logged_task("announce_firmware", self._announce_firmware())` — its failure is logged (not silently swallowed), but it is not restarted.
+
+See the [Connector API spec](../specification/connector#spec-connector-fleetconnector-background-tasks) and the [simple-connector example](https://github.com/inorbit-ai/inorbit-connector-python/blob/main/examples/simple-connector/connector.py).
+
 ### `_disconnect()`
 
 Clean up resources and disconnect from external services. This is called when the connector stops.
diff --git a/examples/simple-connector/connector.py b/examples/simple-connector/connector.py
index 9aa9fa0..560606b 100644
--- a/examples/simple-connector/connector.py
+++ b/examples/simple-connector/connector.py
@@ -57,6 +57,12 @@ async def get_robot_angular_speed() -> float:
     return random.uniform(0.1, 0.9)
 
 
+async def get_robot_firmware() -> str:
+    """Simulate a one-time request to the robot's firmware version API."""
+    await asyncio.sleep(random.uniform(0.1, 0.3))
+    return "fw-2.4.1"
+
+
 class ExampleBotConnector(Connector):
     """The example bot connector.
 
@@ -95,6 +101,44 @@ async def _connect(self) -> None:
         # Do some magic here...
         self._logger.info(f"Connected to robot services at API {self.api_version}")
 
+        # Schedule a background loop that runs at its own cadence (independent of
+        # update_freq) under framework supervision: if it raises, the framework
+        # logs it with a traceback and restarts it, instead of the task dying
+        # silently and freezing this data source.
+        self._create_supervised_task("speed_poll_loop", self._speed_poll_loop)
+
+        # One-shot startup work: fire-and-forget, but a failure is logged (not
+        # lost on an un-awaited task) via _spawn_logged_task. Unlike a
+        # supervised loop, it runs once and is not restarted.
+        self._spawn_logged_task("announce_firmware", self._announce_firmware())
+
+    async def _speed_poll_loop(self) -> None:
+        """Poll the robot's speeds at ~5Hz and publish odometry.
+
+        Demonstrates ``_create_supervised_task``: a long-lived loop running on a
+        faster cadence than ``_execution_loop``.
+        """
+        while True:
+            linear_speed, angular_speed = await asyncio.gather(
+                get_robot_linear_speed(),
+                get_robot_angular_speed(),
+            )
+            self.publish_odometry(
+                linear_speed=linear_speed, angular_speed=angular_speed
+            )
+            await asyncio.sleep(0.2)
+
+    async def _announce_firmware(self) -> None:
+        """Fetch the firmware version once at startup and publish it.
+
+        Demonstrates ``_spawn_logged_task``: a one-shot fire-and-forget task
+        whose failure is logged instead of being silently swallowed on an
+        un-awaited task (but which is not restarted, unlike a supervised loop).
+        """
+        firmware = await get_robot_firmware()
+        self.publish_key_values(firmware_version=firmware)
+        self._logger.info(f"Announced firmware version: {firmware}")
+
     @override
     async def _disconnect(self) -> None:
         """Disconnect from the robot services."""
@@ -135,17 +179,8 @@ async def _execution_loop(self) -> None:
         frame_id = "frameIdA"
         self.publish_pose(x, y, yaw, frame_id)
 
-        # A common pattern is to poll REST endpoints for fresh robot data.
-        # asyncio-compatible libraries are great for this
-        linear_speed, angular_speed = await asyncio.gather(
-            get_robot_linear_speed(),
-            get_robot_angular_speed(),
-        )
-        odometry = {
-            "linear_speed": linear_speed,
-            "angular_speed": angular_speed,
-        }
-        self.publish_odometry(**odometry)
+        # Odometry is polled on a faster cadence in _speed_poll_loop, scheduled
+        # in _connect via _create_supervised_task.
 
         self._logger.info("Robot data updated and published")
 
diff --git a/inorbit_connector/connector.py b/inorbit_connector/connector.py
index c0213bd..1684a68 100644
--- a/inorbit_connector/connector.py
+++ b/inorbit_connector/connector.py
@@ -126,6 +126,10 @@ def __init__(self, config: ConnectorRootConfig, **kwargs) -> None:
         self.__thread = threading.Thread(target=self.__run_loop)
         self.__loop: asyncio.AbstractEventLoop | None = None
 
+        # Long-lived background tasks scheduled via _create_supervised_task /
+        # _spawn_logged_task. Cancelled and awaited in __disconnect().
+        self.__background_tasks: list[asyncio.Task] = []
+
         # Additional initalization arguments
         self.__register_user_scripts = kwargs.get("register_user_scripts", False)
         self.__default_user_scripts_dir = kwargs.get(
@@ -522,6 +526,16 @@ async def __connect(self) -> None:
     async def __disconnect(self) -> None:
         """Disconnect external services and disconnect from InOrbit."""
 
+        # Stop supervised background tasks first so their loops stop touching
+        # robot sessions before we tear those sessions down. Clear the list
+        # up front so the done-callbacks' self-removal is a no-op.
+        tasks = self.__background_tasks
+        self.__background_tasks = []
+        for task in tasks:
+            task.cancel()
+        if tasks:
+            await asyncio.gather(*tasks, return_exceptions=True)
+
         # Tear the fleet down while holding the lock so no robot can be added (via
         # add_robot/update_fleet from another thread) mid-teardown. Free each session
         # from the pool (not just disconnect it) so a subsequent start() rebuilds and
@@ -632,6 +646,102 @@ def _is_session_connected(self, robot_id: str) -> bool:
         except Exception:
             return False
 
+    def _create_supervised_task(
+        self, name: str, coro_factory, restart_delay: float = 5.0
+    ) -> asyncio.Task:
+        """Schedule a long-lived background coroutine under supervision.
+
+        Periodic work placed inside ``_execution_loop`` is already supervised by
+        the framework (``__run_loop`` catches, logs and retries it). Work that
+        needs a cadence other than ``config.update_freq`` (e.g. a fast pose loop
+        and a slower key-value loop) must be scheduled here rather than via a
+        bare ``asyncio.create_task``: a bare task that raises dies silently (its
+        exception is stored on a task nobody awaits and is never logged), is
+        never restarted, and freezes that datasource until the process restarts.
+
+        This wraps ``coro_factory`` in a loop that, if the coroutine returns or
+        raises, logs it (with traceback) and restarts it after ``restart_delay``
+        seconds. ``CancelledError`` is propagated so the task stops cleanly on
+        shutdown. The task is tracked and cancelled in teardown.
+
+        Must be called from the connector's event loop (e.g. from ``_connect``).
+
+        Args:
+            name: Human-readable name used in log messages.
+            coro_factory: Zero-arg callable returning a fresh coroutine; called
+                again on each restart.
+            restart_delay: Seconds to wait before restarting after exit/crash.
+
+        Returns:
+            The supervising ``asyncio.Task``.
+        """
+        task = asyncio.create_task(self.__supervise(name, coro_factory, restart_delay))
+        task.add_done_callback(self.__log_task_exception)
+        self.__background_tasks.append(task)
+        return task
+
+    async def __supervise(self, name, coro_factory, restart_delay) -> None:
+        """Run ``coro_factory`` forever, logging+restarting it on exit/crash."""
+        while True:
+            try:
+                await coro_factory()
+                self._logger.warning(
+                    f"Background task '{name}' exited unexpectedly. "
+                    f"Restarting in {restart_delay}s"
+                )
+            except asyncio.CancelledError:
+                raise
+            except Exception:
+                # TODO(metrics): increment a `background_task_errors` counter
+                # (via inorbit_connector.metrics, alongside execution_loop_errors)
+                # so a crashing/restarting task is visible in monitoring, not
+                # only in the logs.
+                self._logger.exception(
+                    f"Background task '{name}' crashed. "
+                    f"Restarting in {restart_delay}s"
+                )
+            # TODO(backoff): use a capped exponential backoff keyed per task on
+            # consecutive failures instead of a fixed restart_delay, so a task
+            # that fails every iteration (e.g. a persistently malformed upstream
+            # response) doesn't hammer-restart. Reset it after a run stays up for
+            # a while. See the MiR connector's circuit breaker for prior art.
+            await asyncio.sleep(restart_delay)
+
+    def _spawn_logged_task(self, name: str, coro) -> asyncio.Task:
+        """Schedule a one-shot ``coro`` whose failure is logged, not swallowed.
+
+        Use for fire-and-forget tasks that are not long-lived loops. Unlike a
+        bare ``asyncio.create_task``, a failure is logged (with traceback) via a
+        done-callback instead of being stored silently on an un-awaited task.
+        For long-lived loops use ``_create_supervised_task`` instead.
+
+        Args:
+            name: Human-readable name, used in the failure log (mirrors
+                ``_create_supervised_task``'s leading ``name`` argument).
+            coro: The coroutine to run once.
+
+        Returns:
+            The scheduled ``asyncio.Task``.
+        """
+        task = asyncio.create_task(coro, name=name)
+        task.add_done_callback(self.__log_task_exception)
+        self.__background_tasks.append(task)
+        return task
+
+    def __log_task_exception(self, task: asyncio.Task) -> None:
+        """Done-callback: surface a task's exception so it is never silent."""
+        try:
+            self.__background_tasks.remove(task)
+        except ValueError:
+            pass
+        if task.cancelled():
+            return
+        exc = task.exception()
+        if exc is not None:
+            self._logger.error(
+                f"Background task '{task.get_name()}' failed: {exc!r}", exc_info=exc
+            )
+
     def start(self) -> None:
         """Start the connector in a new thread.
 
diff --git a/pyproject.toml b/pyproject.toml
index 4430381..e716ba3 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -35,7 +35,7 @@ classifiers = [
     "License :: OSI Approved :: MIT License",
 ]
 dependencies = [
-    "inorbit-edge[telemetry]>=3.0,<4.0",
+    "inorbit-edge[telemetry]>=3.1.0,<4.0",
     "pydantic>=2.11,<3.0",
     "pydantic-settings>=2.14,<3.0",
     "pytz>=2025.1",
@@ -45,7 +45,7 @@ dependencies = [
 
 [project.optional-dependencies]
 video = [
-    "inorbit-edge[video]>=3.0,<4.0",
+    "inorbit-edge[video]>=3.1.0,<4.0",
 ]
 dev = [
     "bump2version~=1.0",
diff --git a/tests/test_connector.py b/tests/test_connector.py
index a96d314..5061f36 100644
--- a/tests/test_connector.py
+++ b/tests/test_connector.py
@@ -6,6 +6,7 @@
 # SPDX-License-Identifier: MIT
 
 # Standard
+import asyncio
 import logging
 import os
 import threading
@@ -1825,3 +1826,93 @@ class CustomException(Exception):
             execution_status_details="An error occurred executing custom command",
             stderr="CustomException",
         )
+
+
+# ==============================================================================
+# Supervised background tasks
+# ==============================================================================
+
+
+class TestSupervisedBackgroundTasks:
+    """Tests for _create_supervised_task / _spawn_logged_task.
+
+    These guard against the failure mode where a connector schedules periodic
+    work as a bare asyncio.create_task loop: if it raises, the task dies, the
+    exception is stored on a task nobody awaits (never logged), nothing restarts
+    it, and that datasource freezes until the process is restarted.
+    """
+
+    @pytest.fixture(autouse=True)
+    def _concrete_fleet_connector(self):
+        with _concrete(FleetConnector):
+            yield
+
+    @pytest.fixture
+    def connector(self):
+        return FleetConnector(
+            ConnectorRootConfig(
+                api_key="valid_key",
+                connection_config_url=AnyHttpUrl("https://valid.com/"),
+                connector_type="valid_connector",
+                connector_config=DummyConfig(),
+                fleet=[RobotConfig(robot_id="TestRobot1")],
+            )
+        )
+
+    @pytest.mark.asyncio
+    async def test_create_supervised_task_restarts_on_crash(self, connector, caplog):
+        caplog.set_level(logging.ERROR)
+        calls = 0
+
+        async def factory():
+            nonlocal calls
+            calls += 1
+            if calls == 1:
+                raise RuntimeError("boom")
+            await asyncio.Event().wait()  # stay alive once healthy
+
+        task = connector._create_supervised_task("kv-loop", factory, restart_delay=0)
+        try:
+            for _ in range(50):
+                await asyncio.sleep(0.01)
+                if calls >= 2:
+                    break
+        finally:
+            task.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await task
+
+        assert calls >= 2  # restarted after the crash
+        assert any("kv-loop" in r.getMessage() for r in caplog.records)
+
+    @pytest.mark.asyncio
+    async def test_supervised_task_cancelled_on_disconnect(self, connector):
+        started = asyncio.Event()
+
+        async def factory():
+            started.set()
+            await asyncio.Event().wait()
+
+        task = connector._create_supervised_task("loop", factory, restart_delay=0)
+        await asyncio.wait_for(started.wait(), timeout=1.0)
+
+        await connector._FleetConnector__disconnect()
+
+        assert task.done()
+        assert connector._FleetConnector__background_tasks == []
+
+    @pytest.mark.asyncio
+    async def test_spawn_logged_task_logs_failure(self, connector, caplog):
+        caplog.set_level(logging.ERROR)
+
+        async def boom():
+            raise ValueError("kaboom")
+
+        task = connector._spawn_logged_task("oneshot", boom())
+        with pytest.raises(ValueError):
+            await task
+        await asyncio.sleep(0)  # let the done-callback run
+
+        msgs = [r.getMessage() for r in caplog.records]
+        assert any("kaboom" in m for m in msgs)  # the exception is surfaced
+        assert any("oneshot" in m for m in msgs)  # ...tagged with the task name