Three states that looked like health from outside

Audited the engine for what it does when something goes wrong rather than
when it goes right. Each of these left the process healthy, the dashboard
green and the product not working.

A gallery the running encoder cannot read. Embeddings are model-tagged, so
when the fallback chain fires every vector the previous encoder wrote goes
invisible: the shop keeps its customer list and recognises nobody on it,
enrolling each regular a second time. Footfall stays correct, which is why
nothing looks wrong. The only evidence was an INFO line reading 'gallery
ready: 0 embeddings (model w600k_mbf) across 21 identities' - a sentence
that states the disaster and calls it ready. Gallery.health now warns with
the count of PEOPLE lost, not vectors, and carries the same numbers to
/api/stats and /api/health, because a log line on a shop PC is read by
nobody. Proved against the real 87-embedding gallery.

Connected, and sending nothing. 'connected' meant the socket opened, so a
stream that went quiet kept it true while last_frame_age_s climbed and the
heartbeat told head office the camera was up. OpenCV breaks a blocked read
at 30s, but a camera trickling a frame every 20s never trips that and never
recovers. streaming/stalled are reported beside connected and the dashboard
says live/stalled/offline - three states because offline sends you to the
network and stalled says the camera is answering and sending nothing.

The 5-second RTSP timeout that never existed. stimeout;5000000 carried a
comment claiming it bounded a dead camera. Measured on OpenCV 4.11 /
FFmpeg 7.1 against a socket that accepts and then says nothing: 30.0s with
stimeout, 30.0s with timeout, 30.3s with no option at all - identical, so
it was never honoured. stimeout became timeout in FFmpeg 5.0 and neither
reaches the RTSP protocol through this path; the real bound is OpenCV's own
interrupt constant. Replaced by the _tcp_reachable pre-flight probe_source
already used, in code we own: 30.3s -> 0.00-2.02s, each naming its cause.
That matters beyond speed - the VideoCapture constructor is not
interruptible, so stop() could not cut it short and a camera removed from
head office left a daemon thread holding a socket for half a minute.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KGcjxF1cNLcuwc3DAPcnfj

CLAUDE.md

@@ -2285,6 +2285,95 @@ far away, which is the same `fraction_below_gate: 0.59` this file already
 records. The placement is still the limit; the CPU was simply being spent to
 discover that 15 times a second.
 
+## Three states that looked like health from outside
+
+Found by auditing the engine for what it does when something goes wrong,
+rather than when it goes right. Each of these left the process healthy, the
+dashboard green and the product not working - the class of bug this file
+already calls a headcount wrong in a way nobody can detect.
+`tests/test_reliability.py` covers all three.
+
+### A gallery the running encoder cannot read
+
+The fallback chain exists so a memory-starved box still starts, and
+CLAUDE.md already warned that "on a memory-starved box the big model silently
+loses the chain". What it did not say is what that **costs**: embeddings are
+model-tagged, so every vector the previous encoder wrote becomes invisible.
+The shop keeps its whole customer list and recognises nobody on it. Every
+regular is greeted as a stranger and enrolled a second time. Footfall stays
+correct, which is precisely why nothing looks wrong.
+
+The only evidence was an INFO line reading `gallery ready: 0 embeddings
+(model 'w600k_mbf') across 21 identities` - a sentence that says the disaster
+and calls it ready. Run against the real 87-embedding gallery with the
+fallback model forced, it now says:
+
+```
+WARNING gallery: 87 of 87 stored embeddings were written by a DIFFERENT
+        encoder (w600k_r50) and cannot be searched - 21 known people are
+        unrecognisable under the running model 'w600k_mbf'.
+```
+
+`Gallery.health` carries the same numbers to `/api/stats` and
+`gallery_unreadable_embeddings` to `/api/health`, because a log line on a shop
+PC is read by nobody. It travels for the same reason `fraction_below_gate`
+does: beside the number it qualifies. `identities_stranded` is the figure that
+matters - **people lost, not vectors** - and an empty gallery reports zero
+rather than raising an alarm on a fresh install.
+
+### Connected, and sending nothing
+
+`connected` meant *the socket opened*. A stream that opens and then goes quiet
+kept it `true` while `last_frame_age_s` climbed, so the heartbeat told head
+office the camera was up. OpenCV breaks a blocked read after 30s and we
+reconnect - but a camera trickling one frame every 20s never trips that at
+all, so it never reconnects and never recovers.
+
+`stalled()` and `streaming` are reported beside `connected`, and the local
+dashboard now says **live / stalled / offline** rather than live / offline.
+Three states because two of them need opposite actions: offline sends you to
+the network, stalled says the camera is answering and sending nothing. Same
+rule as `artifact` vs `no_faces` in the commissioning verdicts.
+
+`STALL_AFTER_S = 10` is not a preference. The tracker abandons a face after
+`max_misses` (25 frames, ~1.7s at 15 fps), so by 10s every track is long gone
+and 150 frames are missing: whatever this is, recognition cannot use it.
+
+### The 5-second RTSP timeout that never existed
+
+`capture.py` set `stimeout;5000000` with a comment claiming "a 5s socket
+timeout so a dead camera is noticed". Measured against this build (OpenCV
+4.11, FFmpeg 7.1) on a socket that accepts the connection and then says
+nothing:
+
+```
+  stimeout;5000000 -> 30.0s     timeout;5000000 -> 30.0s
+  stimeout;2000000 -> 30.5s     timeout;2000000 -> 30.4s
+  no timeout option at all      -> 30.3s
+```
+
+Identical with the option absent, under either name, so it was never honoured
+through this path - `stimeout` was renamed `timeout` in FFmpeg 5.0 and neither
+reaches the RTSP protocol here. The real bound is OpenCV's own interrupt
+callback, a compile-time constant we do not control. Both names are still set
+(harmless, and right on a build where they do work), but **nothing depends on
+them**.
+
+What replaces it is `_tcp_reachable` in `_open()` - the pre-flight
+`probe_source` already used, in code we own. It matters beyond speed: the
+`cv2.VideoCapture` constructor is not interruptible, so `stop()` could not cut
+it short and a camera removed from head office left a daemon thread holding a
+socket for half a minute. Measured:
+
+```
+  unroutable address     30.3s -> 2.02s   "no response from ... within 2s"
+  host up, port closed   30.3s -> 0.00s   "cannot reach ... Connection refused"
+  wrong port, real cam   30.3s -> 1.01s   "cannot reach ... Connection refused"
+```
+
+`last_error` is reported with the camera, because `connected: false` alone
+cannot tell a wrong IP from a wrong password, and those are different jobs.
+
 ## Setting up on a new machine
 
 1. Copy the `Behavision` folder **including `.env`** (gitignored, holds

behavision/api.py

@@ -167,8 +167,13 @@ def create_app(engine: Engine) -> FastAPI:
     @app.get("/api/health")
     def health() -> dict:
         from .paths import describe
+        # A gallery the running encoder cannot read is the failure most
+        # worth catching from outside: the process is healthy, the cameras
+        # are up, and the shop recognises nobody it already knows.
+        stranded = engine.gallery.health["stranded"]
         return {"status": "ok" if engine.started_at else "starting",
                 "recognition_model": engine.encoder.model_name,
+                "gallery_unreadable_embeddings": stranded,
                 # "where is my database" must be answerable from the API: the
                 # tray, the installer and support all need it, and installed
                 # it is not next to the code.

behavision/capture.py

@@ -20,7 +20,26 @@ log = logging.getLogger(__name__)
 # Set before OpenCV loads ffmpeg, which reads this once.
 #
 # rtsp_transport=tcp: UDP is the default and silently drops frames on lossy
-# Wi-Fi. stimeout: a 5s socket timeout so a dead camera is noticed.
+# Wi-Fi.
+#
+# The timeout here is NOT what bounds a dead camera, and the comment that
+# once said it did was wrong. Measured against OpenCV 4.11 / FFmpeg 7.1 on a
+# socket that accepts the connection and then says nothing:
+#
+#     stimeout;5000000   -> 30.0s      timeout;5000000 -> 30.0s
+#     stimeout;2000000   -> 30.5s      timeout;2000000 -> 30.4s
+#     no timeout option at all         -> 30.3s
+#
+# Identical with the option absent, so it is not being honoured under either
+# name through this path. `stimeout` was renamed `timeout` in FFmpeg 5.0, and
+# neither reaches the RTSP protocol here. What actually bounds it is
+# OpenCV's own interrupt callback (30s for open, 30s for read), which is a
+# compile-time constant we do not control.
+#
+# Both names are still set, because on a build where they DO take effect the
+# shorter bound is what we want and an unrecognised option is ignored. But
+# nothing may depend on it: a wrong address is caught by _tcp_reachable
+# below, in code we own, in under a second.
 #
 # fflags=nobuffer and flags=low_delay: without them ffmpeg's RTSP demuxer
 # holds a comfortable queue of frames before handing over the first, which
@@ -30,10 +49,25 @@ log = logging.getLogger(__name__)
 # reorder wait for the same reason.
 os.environ.setdefault(
     "OPENCV_FFMPEG_CAPTURE_OPTIONS",
-    "rtsp_transport;tcp|stimeout;5000000|fflags;nobuffer|flags;low_delay|max_delay;200000",
+    "rtsp_transport;tcp|stimeout;5000000|timeout;5000000"
+    "|fflags;nobuffer|flags;low_delay|max_delay;200000",
 )
 
 
+# A stream can stay open and stop delivering. OpenCV breaks a blocked read
+# after 30s and we reconnect, but for those 30s `connected` is True and the
+# camera is dead — and a stream that trickles a frame every 20s never trips
+# that timeout at all, so it never reconnects and never recovers either.
+#
+# 10s is not a preference. The tracker gives up on a face after `max_misses`
+# (25 frames, ~1.7s at 15 fps), so by 10s every track is long gone and 150
+# frames are missing: whatever this is, it is not something recognition can
+# work with. Reported separately from `connected` because the two need
+# opposite actions — one says check the network, the other says the camera
+# is answering but sending nothing.
+STALL_AFTER_S = 10.0
+
+
 def _tcp_reachable(source: "str | int", timeout: float
                    ) -> "tuple[bool, str]":
     """Cheap pre-flight for an rtsp:// URL. Non-URL sources pass through."""
@@ -169,6 +203,10 @@ class VideoSource(threading.Thread):
         self.frames_total = 0
         self.reconnects = 0
         self._ever_connected = False
+        # Why the last open failed, in the words an installer can act on.
+        # Without it a camera that never connects reports only `connected:
+        # false`, which cannot distinguish a wrong IP from a wrong password.
+        self.last_error = ""
 
     # -- public ---------------------------------------------------------
     def latest(self) -> "tuple[Optional[np.ndarray], float]":
@@ -199,11 +237,23 @@ class VideoSource(threading.Thread):
     def stop(self) -> None:
         self._stopping.set()
 
+    def stalled(self) -> bool:
+        """Open, but not delivering. See STALL_AFTER_S."""
+        if not self.connected or not self._frame_ts:
+            return False
+        return (time.time() - self._frame_ts) > STALL_AFTER_S
+
     def stats(self) -> dict:
         return {
             "camera_id": self.camera_id,
             "url": self._display_url,
             "connected": self.connected,
+            # Connected AND delivering. `connected` alone stays true through
+            # a stall, so it is the wrong thing for a dashboard to colour a
+            # camera green on.
+            "streaming": self.connected and not self.stalled(),
+            "stalled": self.stalled(),
+            "last_error": self.last_error,
             "frames_total": self.frames_total,
             "reconnects": self.reconnects,
             "last_frame_age_s": round(time.time() - self._frame_ts, 1)
@@ -260,6 +310,19 @@ class VideoSource(threading.Thread):
         log.info("[%s] capture stopped", self.camera_id)
 
     def _open(self) -> Optional[cv2.VideoCapture]:
+        # Pre-flight the socket, exactly as probe_source does. Without it a
+        # camera that is off, moved or mistyped costs 30s per attempt inside
+        # the VideoCapture constructor (measured; it is OpenCV's interrupt
+        # timeout, not ours to shorten) — and the constructor is not
+        # interruptible, so stop() cannot cut it short and a removed camera
+        # leaves a daemon thread holding a socket for half a minute. A
+        # refused or unroutable address answers in well under a second, which
+        # is also what lets the backoff below mean what it says.
+        reachable, why = _tcp_reachable(self._source, 2.0)
+        if not reachable:
+            log.debug("[%s] %s", self.camera_id, why)
+            self.last_error = why
+            return None
         try:
             if isinstance(self._source, int):
                 cap = cv2.VideoCapture(self._source)
@@ -268,8 +331,12 @@ class VideoSource(threading.Thread):
             cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
             if not cap.isOpened():
                 cap.release()
+                self.last_error = ("reachable, but the stream would not open "
+                                   "- check the path and credentials")
                 return None
+            self.last_error = ""
             return cap
         except cv2.error:
             log.exception("[%s] VideoCapture error", self.camera_id)
+            self.last_error = "VideoCapture error - see the engine log"
             return None

behavision/engine.py

@@ -18,6 +18,7 @@ import numpy as np
 
 from .attributes import AttributeEstimator, aggregate as aggregate_attrs
 from .cameras import CameraStore
+from . import capture
 from .capture import VideoSource
 from .faces import FaceOutbox
 from .commission import CommissionRun
@@ -169,6 +170,7 @@ class CameraWorker(threading.Thread):
         self._overlay_ts = 0.0
         self._last_frame_ts = 0.0
         self._was_connected = False
+        self._was_stalled = False
         self.frames_processed = 0
         # Motion gate state: a 160x90 greyscale thumbnail of the last frame we
         # actually searched, and how many frames we have skipped since.
@@ -346,6 +348,20 @@ class CameraWorker(threading.Thread):
                 type="camera.up" if connected else "camera.down",
                 camera_id=self.cam_cfg.id))
 
+        # A stall is not a disconnect and must not be reported as one: the
+        # socket is fine, the camera is answering, and nothing is arriving.
+        # Logged on the transition only — a per-frame warning would bury the
+        # one line that matters under thousands of copies of itself.
+        stalled = self.source.stalled()
+        if stalled != self._was_stalled:
+            self._was_stalled = stalled
+            if stalled:
+                log.warning("[%s] connected but no frame for over %.0fs - the "
+                            "camera is answering and sending nothing",
+                            self.cam_cfg.id, capture.STALL_AFTER_S)
+            else:
+                log.info("[%s] frames resumed", self.cam_cfg.id)
+
     def _finish_track(self, track: Track, ts: float) -> None:
         """Record what became of a track, once, as it ends.
 
@@ -681,7 +697,10 @@ class Engine:
                 "age_model": ("genderage" if self.attributes is not None
                               and self.attributes.has_genderage else "caffe/none"),
             },
-            "gallery": self.store.stats(),
+            # Counts, plus whether the running encoder can actually SEARCH
+            # them. A gallery of 21 identities that the loaded model cannot
+            # read is the silent version of an empty one.
+            "gallery": {**self.store.stats(), **self.gallery.health},
             "cameras": [w.stats() for w in self.snapshot_workers()],
         }
 

behavision/gallery/service.py

@@ -53,10 +53,56 @@ class Gallery:
         # vectors from a different model are numerically incompatible.
         ids, vecs = store.all_embeddings(index.dim, model=model_name)
         index.add(ids, vecs)
+        self.health = self._assess(len(ids))
+        if self.health["stranded"]:
+            # Not an INFO line. The encoder fallback chain exists so a
+            # memory-starved box still runs, and when it fires every vector
+            # written by the previous encoder becomes invisible: the shop
+            # keeps its customer list and recognises nobody on it, greeting
+            # every regular as new and enrolling them a second time. Footfall
+            # stays right, which is exactly why nothing looks wrong. The old
+            # message for that state was "gallery ready: 0 embeddings".
+            log.warning(
+                "gallery: %d of %d stored embeddings were written by a "
+                "DIFFERENT encoder (%s) and cannot be searched - %d known "
+                "%s unrecognisable under the running model '%s'. Either "
+                "restore that model or accept that these identities start "
+                "over.",
+                self.health["stranded"], self.health["stored"],
+                ", ".join(sorted(self.health["other_models"])),
+                self.health["identities_stranded"],
+                "person is" if self.health["identities_stranded"] == 1
+                else "people are",
+                model_name)
         log.info("gallery ready: %d embeddings (model '%s') across %d "
                  "identities", len(ids), model_name,
                  store.stats()["identities"])
 
+    def _assess(self, usable: int) -> dict:
+        """What share of the gallery the running encoder can actually reach.
+
+        Reported rather than merely logged, because a log line on a shop PC
+        is read by nobody: this travels to head office the same way
+        `fraction_below_gate` does, beside the number it qualifies.
+        """
+        counts = self.store.model_counts()
+        stored = sum(counts.values())
+        others = {m: n for m, n in counts.items() if m != self.model_name}
+        identities = self.store.stats()["identities"]
+        return {
+            "model": self.model_name,
+            "stored": stored,
+            "usable": usable,
+            "stranded": sum(others.values()),
+            "other_models": sorted(others),
+            "identities": identities,
+            "identities_usable": self.store.identities_with_model(
+                self.model_name),
+            "identities_stranded": max(
+                0, identities - self.store.identities_with_model(
+                    self.model_name)),
+        }
+
     def resolve(self, embedding: np.ndarray, quality: float, camera_id: str,
                 ts: "float | None" = None,
                 attributes: "dict | None" = None,

behavision/gallery/store.py

@@ -281,6 +281,31 @@ class IdentityStore:
             return None
         return np.frombuffer(row["vector"], dtype=np.float32), float(row["quality"])
 
+    def model_counts(self) -> "dict[str, int]":
+        """How many stored embeddings each encoder produced.
+
+        The gallery only ever searches vectors tagged with the *running*
+        encoder, so this is what says whether the rest of the gallery is
+        reachable at all. See `Gallery.health` for why that matters.
+        """
+        with self._lock:
+            rows = self._db.execute(
+                "SELECT model, COUNT(*) AS n FROM embeddings "
+                "GROUP BY model").fetchall()
+        return {str(r["model"]): int(r["n"]) for r in rows}
+
+    def identities_with_model(self, model: str) -> int:
+        """Identities holding at least one embedding from this encoder.
+
+        Not the same as the identity count: an identity whose only vectors
+        came from a previous encoder still exists, and is unrecognisable.
+        """
+        with self._lock:
+            row = self._db.execute(
+                "SELECT COUNT(DISTINCT identity_id) AS n FROM embeddings "
+                "WHERE model=?", (model,)).fetchone()
+        return int(row["n"]) if row else 0
+
     def embedding_owners(self, model: "str | None" = None) -> "dict[int, int]":
         """embedding_id -> identity_id, for turning index hits into identity
         pairs without a round trip to SQLite per hit."""

behavision/static/dashboard.html

@@ -514,11 +514,24 @@ async function refresh() {
     ]);
     renderFeeds(camList);
     renderCameras(camList);
+    // 'stalled' is its own word on purpose: connected and offline send you
+    // to the network, a camera that is answering and sending nothing does
+    // not. Three states, because two of them need opposite actions.
     const cams = stats.cameras.map(c =>
-      `${c.camera_id}: ${c.connected ? 'live' : 'offline'}`).join(' · ');
+      `${c.camera_id}: ${c.streaming ? 'live' : c.connected ? 'stalled' : 'offline'}`
+    ).join(' · ');
+    // The one failure that otherwise looks like perfect health: the encoder
+    // that loaded cannot read the embeddings already stored, so every known
+    // customer is a stranger. Counts stay right, which is why it needs saying.
+    const stranded = stats.gallery.stranded || 0;
+    const warn = stranded
+      ? ` · ⚠ ${stats.gallery.identities_stranded} people unrecognisable `
+        + `(${stranded} embeddings from ${stats.gallery.other_models.join(', ')}, `
+        + `running ${stats.gallery.model})`
+      : '';
     // textContent, not innerHTML — no escaping needed here.
     document.getElementById('status').textContent =
-      `${cams} · ${stats.gallery.identities} people · ${stats.gallery.sightings} sightings`;
+      `${cams} · ${stats.gallery.identities} people · ${stats.gallery.sightings} sightings${warn}`;
 
     document.getElementById('events').innerHTML = events.map(e => {
       const cls = e.type === 'person.new' ? 'new'

tests/test_reliability.py

@@ -0,0 +1,170 @@
+"""Failure modes that look like health from outside.
+
+Each of these was a state the engine could be in while every existing test
+passed and the dashboard showed green. They are grouped because they share
+one property: the process is fine and the product is not working.
+"""
+from __future__ import annotations
+
+import time
+
+import numpy as np
+import pytest
+
+from behavision import capture
+from behavision.capture import VideoSource
+from behavision.config import RecognitionSection
+from behavision.gallery import Gallery, IdentityStore, VectorIndex
+from behavision.recognition import EMBEDDING_DIM
+
+
+def _vec(seed: int) -> np.ndarray:
+    """A distinct unit vector per seed.
+
+    Orthogonal per index, NOT a constant fill: a vector of all 0.3 and one of
+    all 0.6 normalise to the same direction, so a fixture built that way would
+    call two 'different' people one identity and prove nothing.
+    """
+    v = np.zeros(EMBEDDING_DIM, dtype=np.float32)
+    v[seed % EMBEDDING_DIM] = 1.0
+    return v
+
+
+def _gallery(store: IdentityStore, model: str) -> Gallery:
+    return Gallery(store, VectorIndex(EMBEDDING_DIM), RecognitionSection(),
+                   model_name=model)
+
+
+# -- the gallery the running encoder cannot read ------------------------
+
+def test_matching_model_is_fully_usable(tmp_path):
+    store = IdentityStore(tmp_path / "g.db")
+    ident = store.create_identity("Alice")
+    store.add_embedding(ident, _vec(1), 0.8, "w600k_r50")
+
+    health = _gallery(store, "w600k_r50").health
+    assert health["usable"] == 1
+    assert health["stranded"] == 0
+    assert health["identities_stranded"] == 0
+
+
+def test_fallback_encoder_strands_the_gallery_and_says_so(tmp_path, caplog):
+    """The whole point: 2 known people, 0 recognisable, and it must be LOUD.
+
+    This is what a memory-starved box does when the 166 MB model loses the
+    fallback chain to the 13 MB one. Footfall keeps counting, so nothing
+    downstream looks wrong; every regular is simply greeted as a stranger and
+    enrolled a second time.
+    """
+    store = IdentityStore(tmp_path / "g.db")
+    for i in (1, 2):
+        ident = store.create_identity(f"Person {i}")
+        store.add_embedding(ident, _vec(i), 0.8, "w600k_r50")
+
+    with caplog.at_level("WARNING"):
+        health = _gallery(store, "w600k_mbf").health
+
+    assert health["usable"] == 0
+    assert health["stranded"] == 2
+    assert health["identities_stranded"] == 2
+    assert health["other_models"] == ["w600k_r50"]
+
+    warning = " ".join(r.getMessage() for r in caplog.records
+                       if r.levelname == "WARNING")
+    assert "w600k_r50" in warning and "w600k_mbf" in warning, warning
+
+
+def test_partially_stranded_counts_only_the_unreachable(tmp_path):
+    """A mixed gallery is the normal state after a model change, and the
+    number that matters is how many people are lost, not how many vectors."""
+    store = IdentityStore(tmp_path / "g.db")
+    old = store.create_identity("Old")
+    store.add_embedding(old, _vec(1), 0.8, "w600k_mbf")
+    store.add_embedding(old, _vec(2), 0.8, "w600k_mbf")
+    both = store.create_identity("Both")
+    store.add_embedding(both, _vec(3), 0.8, "w600k_mbf")
+    store.add_embedding(both, _vec(4), 0.8, "w600k_r50")
+
+    health = _gallery(store, "w600k_r50").health
+    assert health["stored"] == 4
+    assert health["usable"] == 1
+    assert health["stranded"] == 3
+    # "Both" survives the change; only "Old" is unrecognisable.
+    assert health["identities_stranded"] == 1
+
+
+def test_empty_gallery_is_not_reported_as_stranded(tmp_path):
+    """A new install must not raise an alarm about a gallery nobody has
+    filled yet - crying wolf here trains people to ignore the real one."""
+    health = _gallery(IdentityStore(tmp_path / "g.db"), "w600k_r50").health
+    assert health["stranded"] == 0
+    assert health["identities_stranded"] == 0
+
+
+# -- open, but not delivering -------------------------------------------
+
+def _source() -> VideoSource:
+    return VideoSource("cam", "rtsp://198.51.100.9:554/x")
+
+
+def test_a_camera_that_never_connected_is_not_stalled():
+    """`stalled` must mean 'was working, stopped'. A camera that has never
+    delivered a frame is a different fault with a different fix."""
+    src = _source()
+    assert src.stalled() is False
+    src.connected = True
+    assert src.stalled() is False, "no frame ever seen is not a stall"
+
+
+def test_a_fresh_frame_is_not_a_stall():
+    src = _source()
+    src.connected = True
+    src._frame_ts = time.time()
+    assert src.stalled() is False
+    assert src.stats()["streaming"] is True
+
+
+def test_an_old_frame_on_an_open_socket_is_a_stall():
+    src = _source()
+    src.connected = True
+    src._frame_ts = time.time() - (capture.STALL_AFTER_S + 1)
+    assert src.stalled() is True
+    stats = src.stats()
+    # The distinction that matters: still connected, no longer streaming.
+    assert stats["connected"] is True
+    assert stats["streaming"] is False
+    assert stats["stalled"] is True
+
+
+def test_a_disconnected_camera_is_reported_as_down_not_stalled():
+    """Two states, opposite actions: check the network vs. the camera is
+    answering and sending nothing. They must never share a verdict."""
+    src = _source()
+    src.connected = False
+    src._frame_ts = time.time() - 3600
+    assert src.stalled() is False
+    assert src.stats()["streaming"] is False
+
+
+# -- a wrong address must not cost 30 seconds ---------------------------
+
+def test_unreachable_source_fails_fast_with_a_reason():
+    """_open() used to hand an unroutable address straight to OpenCV, which
+    blocks ~30s inside the constructor and cannot be interrupted by stop().
+    198.51.100.0/24 is TEST-NET-2 and routes nowhere.
+    """
+    src = VideoSource("cam", "rtsp://198.51.100.9:554/x")
+    started = time.time()
+    assert src._open() is None
+    elapsed = time.time() - started
+    assert elapsed < 8.0, f"pre-flight took {elapsed:.1f}s"
+    assert src.last_error, "a failed open must say why"
+    assert "198.51.100.9" in src.last_error
+    assert src.stats()["last_error"] == src.last_error
+
+
+def test_webcam_sources_skip_the_preflight():
+    """An int source is a local device with no host to reach; the check must
+    pass it through rather than refuse it."""
+    ok, why = capture._tcp_reachable(0, 1.0)
+    assert ok is True and why == ""