fix(camera): share one connection between concurrent one-shot captures (#2705)

Bambu firmware allows exactly one camera connection. The existing guards
(is_stream_active / try_get_active_buffered_frame, #1271 and #1348) only
stop a one-shot capturer from competing with the fan-out broadcaster.
Nothing coordinated the capturers with each other, so with no viewer
attached every consumer correctly concluded it was not competing with a
viewer and then collided with the others. On the reporter's P2S an Obico
poll and a snapshot opened two RTSP sockets 207 ms apart, which knocked
over the fan-out stream feeding the camera wall; it was then reaped for
having received no frames for 58s.

capture_camera_frame_bytes() now coalesces: the first caller opens the
connection, callers arriving while it is in flight await the same result.
Eight paths reach that function independently - Obico polling, the
snapshot route, the finish-photo moment and its disk-writing sibling,
plate detection, the camera test and the diagnose tool - so the
single-flight sits at the bottom of the stack and no call site changes.

Keyed by IP, since that is what the firmware's limit applies to and the
function never sees a printer_id. The key excludes the timeout on
purpose: the call sites disagree about it, from 10s to 30s, so keying on
it would mean the Obico-vs-snapshot pair from the report never coalesced
at all.

It coalesces, it does not cache. A call arriving after the previous
capture finished still captures fresh, because plate detection and the
finish-photo path judge a running print from these frames and a stale one
there is worse than a slow one - #1397 was a finish photo taken seconds
late showing the bed already lowered.

Each caller waits on its own deadline rather than inheriting whichever
one happened to open the connection, and shield() means giving up leaves
the capture running for whoever else is still waiting. A follower whose
leader fails takes a turn of its own instead of inheriting a failure it
never had a chance to avoid; the leader has finished by then, so there is
nothing left to compete with. Bounded at two rounds. That also covers the
follower whose timeout is longer than the leader's, which coalescing
alone cannot. Cancellation is disambiguated via leader.cancelled(), so a
follower's own cancellation propagates while a cancelled leader is
treated as a failed one.

The leader is deliberately not wrapped in a second wait_for: the
implementation already enforces the timeout internally, where it can also
kill the ffmpeg process, and an outer deadline would abandon the
subprocess instead of killing it.

The diagnose tool now marks a stage whose frame came from a capture
already in flight as coalesced_capture. The pass is real evidence the
camera works, but duration_ms is then mostly time spent queueing, and a
diagnostic must not report a connection it never opened - the same reason
that file declares its live_stream_active shortcut instead of quietly
passing. Failures are not annotated, since a follower whose leader fails
goes on to capture on its own.

backend/app/services/camera.py

@@ -6,6 +6,7 @@ Supports two camera protocols:
 """
 
 import asyncio
+import functools
 import logging
 import os
 import shutil
@@ -34,6 +35,26 @@ _rtsp_socket_timeout_flag: str | None = None
 # The cleanup task in routes/camera.py checks this set to avoid killing active captures.
 _active_capture_pids: set[int] = set()
 
+# In-flight one-shot captures, keyed by printer IP (#2705).
+#
+# Bambu firmware allows exactly one camera connection, and the existing guards
+# (is_stream_active / try_get_active_buffered_frame, #1271 + #1348) only stop a
+# capturer from competing with the fan-out BROADCASTER. They do nothing for
+# capturer-vs-capturer with no viewer attached, where every consumer correctly
+# concludes it isn't competing with a viewer and then collides with the others.
+# Eight paths reach capture_camera_frame_bytes() independently — Obico polling,
+# /camera/snapshot, the finish-photo moment and its disk-writing sibling, plate
+# detection, the camera test and the diagnose tool — so the single-flight lives
+# at the bottom of the stack and needs no call-site changes.
+#
+# Keyed by IP rather than printer_id because IP is what the firmware's one-
+# connection limit applies to: two printer rows pointing at the same address
+# still share one camera. (This function never sees a printer_id anyway.) The
+# key deliberately excludes the timeout, or callers that disagree about it —
+# and they all do, from 10s to 30s — would never coalesce, which is exactly
+# the Obico-vs-snapshot pair from the report.
+_inflight_captures: dict[str, asyncio.Task[bytes | None]] = {}
+
 
 def get_ffmpeg_path() -> str | None:
     """Find the ffmpeg executable path.
@@ -529,6 +550,38 @@ async def capture_camera_frame(
     return False
 
 
+def capture_in_flight(ip_address: str) -> bool:
+    """Return True iff a one-shot capture for this IP is running right now.
+
+    For callers that need to know whether they will JOIN someone else's
+    capture rather than perform their own — currently only the diagnose tool,
+    which reports on what it measured and so must not present a coalesced
+    frame as proof that it opened its own connection (see camera_diagnose).
+
+    Ordinary consumers should ignore this: they want "a recent frame", and
+    capture_camera_frame_bytes() already does the right thing for them.
+    """
+    task = _inflight_captures.get(ip_address)
+    return task is not None and not task.done()
+
+
+def _discard_inflight_capture(ip_address: str, task: asyncio.Task) -> None:
+    """Done-callback: drop the finished task from the in-flight registry.
+
+    Guarded on identity so a slow task that finishes after a newer capture
+    has registered can't evict its successor.
+
+    Also retrieves the exception, if any. The leader normally awaits the task
+    and would surface it, but a leader whose own caller was cancelled leaves
+    nobody to collect it — and an unretrieved task exception is logged by
+    asyncio as a warning with a traceback at an arbitrary later point.
+    """
+    if _inflight_captures.get(ip_address) is task:
+        del _inflight_captures[ip_address]
+    if not task.cancelled() and task.exception() is not None:
+        logger.debug("In-flight camera capture for %s ended in an exception", ip_address)
+
+
 async def capture_camera_frame_bytes(
     ip_address: str,
     access_code: str,
@@ -537,18 +590,95 @@ async def capture_camera_frame_bytes(
 ) -> bytes | None:
     """Capture a single frame and return as JPEG bytes (no disk write).
 
-    Uses the same protocol selection as capture_camera_frame but returns
-    bytes directly instead of writing to disk.
+    Concurrent callers for the same printer share one capture (#2705): the
+    first opens the connection, everyone arriving while it is in flight awaits
+    the same result. Every consumer here wants "a recent frame" rather than
+    "a frame captured at exactly my timestamp", so handing identical bytes to
+    simultaneous callers is correct — and it is the only way to honour the
+    firmware's one-connection limit without serialising captures behind a lock
+    (which would just turn a collision into a queue).
+
+    This coalesces; it does not cache. A call that arrives after the previous
+    capture finished always captures fresh. Two consumers of these frames —
+    plate detection and the finish-photo path — decide things about a running
+    print from them, and a stale frame there is worse than a slow one: the
+    whole of #1397 was a finish photo taken seconds late showing the bed
+    already lowered.
 
     Args:
         ip_address: Printer IP address
         access_code: Printer access code
         model: Printer model (X1, H2D, P1, A1, etc.)
-        timeout: Timeout in seconds for the capture operation
+        timeout: Timeout in seconds for the capture operation. Applies to this
+            caller's own wait, including when it joins another caller's
+            capture — the call sites disagree about the value (10s for plate
+            detection, 20s for Obico), and a follower must not silently
+            inherit the leader's deadline in either direction.
 
     Returns:
         JPEG bytes if capture was successful, None otherwise
     """
+    # A follower whose leader fails takes a turn of its own rather than
+    # inheriting a failure it never had a chance to avoid — by then the leader
+    # has finished, so there is no socket left to compete with. Bounded at two
+    # rounds: if the capture we joined AND its replacement both failed, a third
+    # connection won't help, and this caller has already spent its patience.
+    for _ in range(2):
+        leader = _inflight_captures.get(ip_address)
+        if leader is None or leader.done():
+            break
+        try:
+            frame = await asyncio.wait_for(asyncio.shield(leader), timeout=timeout)
+        except TimeoutError:
+            # shield() keeps the capture running for whoever else is still
+            # waiting on it — giving up is this caller's decision alone.
+            logger.warning(
+                "Gave up waiting %ss on the in-flight camera capture for %s",
+                timeout,
+                ip_address,
+            )
+            return None
+        except asyncio.CancelledError:
+            # Distinguish "the capture I joined was cancelled" from "I was
+            # cancelled". Only the former is ours to recover from.
+            if not leader.cancelled():
+                raise
+            logger.info("In-flight camera capture for %s was cancelled; capturing our own", ip_address)
+            continue
+        if frame is not None:
+            logger.info(
+                "Reusing in-flight camera capture for %s: %s bytes (no second connection opened)",
+                ip_address,
+                len(frame),
+            )
+            return frame
+        logger.info("In-flight camera capture for %s failed; capturing our own", ip_address)
+    else:
+        return None
+
+    task = asyncio.create_task(_capture_camera_frame_bytes_uncoalesced(ip_address, access_code, model, timeout))
+    _inflight_captures[ip_address] = task
+    task.add_done_callback(functools.partial(_discard_inflight_capture, ip_address))
+    # No wait_for here: this caller IS the capture, and the implementation
+    # already enforces `timeout` internally where it can also kill the ffmpeg
+    # process. A second deadline on top would abandon the subprocess instead.
+    # shield() so that a cancelled leader (a client navigating away mid-
+    # snapshot is routine) doesn't take the capture down with it — the
+    # followers already waiting on it still get their frame.
+    return await asyncio.shield(task)
+
+
+async def _capture_camera_frame_bytes_uncoalesced(
+    ip_address: str,
+    access_code: str,
+    model: str | None,
+    timeout: int = 15,
+) -> bytes | None:
+    """Open a connection and capture one frame. See capture_camera_frame_bytes.
+
+    Callers want that wrapper, not this: it opens a socket unconditionally,
+    which is the collision #2705 is about.
+    """
     # Chamber image models: A1/P1 - returns bytes directly
     if is_chamber_image_model(model):
         logger.info("Capturing camera frame bytes from %s using chamber image protocol (model: %s)", ip_address, model)

backend/app/services/camera_diagnose.py

@@ -35,6 +35,13 @@ out broadcaster to prevent). When ``is_stream_active`` reports True
 AND a buffered frame is fresh (last 10 s), we short-circuit the test
 with ``live_stream_active`` and report success — the user is
 literally watching the camera right now, no test needed.
+
+The related case is another one-shot capture (Obico polling, the cam
+wall) being in flight when the user hits Diagnose. There the capture
+layer coalesces for us (#2705) and no competing socket is opened, but
+the frame we get back was someone else's — so ``first_frame`` still
+passes and carries a ``coalesced_capture`` code, because a diagnostic
+that reports a connection it didn't open is worse than a slow one.
 """
 
 from __future__ import annotations
@@ -46,6 +53,7 @@ from dataclasses import dataclass, field
 
 from backend.app.services.camera import (
     capture_camera_frame_bytes,
+    capture_in_flight,
     get_camera_port,
     is_chamber_image_model,
 )
@@ -69,8 +77,10 @@ class CameraDiagnoseStage:
     name: str  # "tcp_reachable" | "first_frame" | "live_stream_active"
     status: str  # "ok" | "failed" | "skipped"
     duration_ms: int = 0
-    # Optional machine-readable code for failures so the frontend can
-    # render a stage-specific hint without parsing free-text errors.
+    # Optional machine-readable code so the frontend can render a stage-
+    # specific hint without parsing free-text errors. Usually a failure
+    # reason; "coalesced_capture" qualifies a PASS whose frame came from a
+    # capture already in flight, so duration_ms isn't a connection time.
     code: str | None = None
 
 
@@ -166,6 +176,15 @@ async def _check_first_frame(
     """Stage 2 — capture one frame end-to-end. Combines auth + protocol
     handshake + first keyframe; either it works or it doesn't."""
     started = time.monotonic()
+    # A capture already running for this printer (an Obico poll, the cam wall)
+    # means capture_camera_frame_bytes will hand us THAT capture's frame rather
+    # than opening its own connection (#2705). Good for the printer, but this
+    # stage exists to report what it measured: the frame would be real evidence
+    # the camera works, while duration_ms would be mostly time spent queueing,
+    # and a pass would be claimed for a connection we never opened. So the
+    # stage says so, the same way the live-stream shortcut above declares
+    # itself instead of quietly passing.
+    coalesced = capture_in_flight(ip_address)
     try:
         jpeg = await capture_camera_frame_bytes(
             ip_address=ip_address,
@@ -190,7 +209,11 @@ async def _check_first_frame(
             name="first_frame",
             status="ok",
             duration_ms=int((time.monotonic() - started) * 1000),
+            code="coalesced_capture" if coalesced else None,
         )
+    # No annotation on the failure path: a follower whose leader fails goes on
+    # to capture on its own, so a None here means this stage did get its own
+    # attempt (or watched two consecutive captures fail — same verdict).
     return CameraDiagnoseStage(
         name="first_frame",
         status="failed",

backend/tests/unit/services/test_camera_capture_coalescing.py

@@ -0,0 +1,288 @@
+"""Single-flight coalescing of one-shot camera captures (#2705).
+
+Bambu firmware allows exactly one camera connection. The pre-existing guards
+(``is_stream_active`` / ``try_get_active_buffered_frame``, #1271 + #1348) only
+keep a one-shot capturer from competing with the fan-out broadcaster; nothing
+kept the capturers from competing with EACH OTHER when no viewer was attached,
+so an Obico poll and a ``/camera/snapshot`` 200 ms apart each opened their own
+RTSP socket and knocked the other over.
+
+These tests drive ``capture_camera_frame_bytes`` at the public boundary and
+count how many times the underlying capture ran, since "how many connections
+did we open" is the entire point of the fix.
+"""
+
+import asyncio
+
+import pytest
+
+from backend.app.services import camera as camera_module
+from backend.app.services.camera import capture_camera_frame_bytes, capture_in_flight
+
+FRAME_A = b"\xff\xd8" + b"a" * 200 + b"\xff\xd9"
+FRAME_B = b"\xff\xd8" + b"b" * 200 + b"\xff\xd9"
+
+
+@pytest.fixture(autouse=True)
+def _clear_inflight():
+    """The registry is module-global; don't leak tasks between tests."""
+    camera_module._inflight_captures.clear()
+    yield
+    camera_module._inflight_captures.clear()
+
+
+class RecordingCapture:
+    """Stand-in for the real capture, recording each call.
+
+    ``gate`` (when set) holds every capture open until released, which is how
+    these tests create the overlap window that used to produce two sockets.
+    """
+
+    def __init__(self, frames=(FRAME_A, FRAME_B), gate: asyncio.Event | None = None):
+        self.calls: list[tuple[str, int]] = []
+        self._frames = list(frames)
+        self._gate = gate
+        self.started = asyncio.Event()
+
+    async def __call__(self, ip_address, access_code, model, timeout=15):
+        self.calls.append((ip_address, timeout))
+        self.started.set()
+        if self._gate is not None:
+            await self._gate.wait()
+        return self._frames.pop(0) if self._frames else None
+
+    @property
+    def count(self) -> int:
+        return len(self.calls)
+
+
+@pytest.fixture
+def patch_capture(monkeypatch):
+    def _install(capture):
+        monkeypatch.setattr(camera_module, "_capture_camera_frame_bytes_uncoalesced", capture)
+        return capture
+
+    return _install
+
+
+async def _let_leader_start(capture: RecordingCapture) -> None:
+    """Wait until the leader is inside the capture, so the next caller joins it.
+
+    Without this the second caller can reach the registry before the first has
+    even been scheduled, which tests a different (and uninteresting) race.
+    """
+    await asyncio.wait_for(capture.started.wait(), timeout=1)
+
+
+@pytest.mark.asyncio
+async def test_simultaneous_callers_share_one_capture(patch_capture):
+    """The reported collision: two consumers, one connection, two frames."""
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(gate=gate))
+
+    leader = asyncio.create_task(capture_camera_frame_bytes("10.0.2.43", "code", "P2S", timeout=20))
+    await _let_leader_start(capture)
+    follower = asyncio.create_task(capture_camera_frame_bytes("10.0.2.43", "code", "P2S", timeout=15))
+    await asyncio.sleep(0)
+    gate.set()
+
+    assert await leader == FRAME_A
+    assert await follower == FRAME_A
+    assert capture.count == 1
+
+
+@pytest.mark.asyncio
+async def test_five_callers_one_capture(patch_capture):
+    """Verified on live hardware in the report: 5 callers, 1 connection."""
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(gate=gate))
+
+    first = asyncio.create_task(capture_camera_frame_bytes("10.0.2.43", "code", "P2S"))
+    await _let_leader_start(capture)
+    rest = [asyncio.create_task(capture_camera_frame_bytes("10.0.2.43", "code", "P2S")) for _ in range(4)]
+    await asyncio.sleep(0)
+    gate.set()
+
+    assert await asyncio.gather(first, *rest) == [FRAME_A] * 5
+    assert capture.count == 1
+
+
+@pytest.mark.asyncio
+async def test_different_printers_do_not_coalesce(patch_capture):
+    """The one-connection limit is per printer, so the key must be too."""
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(gate=gate))
+
+    one = asyncio.create_task(capture_camera_frame_bytes("10.0.2.43", "code", "P2S"))
+    await _let_leader_start(capture)
+    two = asyncio.create_task(capture_camera_frame_bytes("10.0.2.44", "code", "P2S"))
+    await asyncio.sleep(0)
+    gate.set()
+
+    assert {await one, await two} == {FRAME_A, FRAME_B}
+    assert capture.count == 2
+    assert {ip for ip, _ in capture.calls} == {"10.0.2.43", "10.0.2.44"}
+
+
+@pytest.mark.asyncio
+async def test_coalescing_is_not_caching(patch_capture):
+    """Sequential callers each capture fresh.
+
+    Deliberate: plate detection and the finish-photo path decide things about a
+    running print from these frames, and #1397 was a finish photo a few seconds
+    stale showing the bed already lowered.
+    """
+    capture = patch_capture(RecordingCapture())
+
+    assert await capture_camera_frame_bytes("10.0.2.43", "code", "P2S") == FRAME_A
+    assert await capture_camera_frame_bytes("10.0.2.43", "code", "P2S") == FRAME_B
+    assert capture.count == 2
+
+
+@pytest.mark.asyncio
+async def test_registry_is_empty_after_a_capture_finishes(patch_capture):
+    """No leak, and nothing left behind for the next caller to join."""
+    patch_capture(RecordingCapture())
+
+    await capture_camera_frame_bytes("10.0.2.43", "code", "P2S")
+    await asyncio.sleep(0)  # let the done-callback run
+
+    assert camera_module._inflight_captures == {}
+    assert capture_in_flight("10.0.2.43") is False
+
+
+@pytest.mark.asyncio
+async def test_failed_leader_does_not_poison_its_followers(patch_capture):
+    """A follower that never got its own attempt gets one when the leader fails.
+
+    Safe by then: the leader has finished, so there is no socket to compete
+    with. This also covers the follower whose timeout is LONGER than the
+    leader's — it isn't cut short by someone else's deadline.
+    """
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(frames=(None, FRAME_B), gate=gate))
+
+    leader = asyncio.create_task(capture_camera_frame_bytes("10.0.2.43", "code", "P2S", timeout=10))
+    await _let_leader_start(capture)
+    follower = asyncio.create_task(capture_camera_frame_bytes("10.0.2.43", "code", "P2S", timeout=20))
+    await asyncio.sleep(0)
+    gate.set()
+
+    assert await leader is None
+    assert await follower == FRAME_B
+    assert capture.count == 2
+
+
+@pytest.mark.asyncio
+async def test_two_consecutive_failures_give_up(patch_capture):
+    """Bounded retry: a follower doesn't chase failing captures forever.
+
+    Two followers behind a failing leader. The first takes its own turn, the
+    second joins THAT capture, and when it fails too the second gives up rather
+    than opening a third connection.
+    """
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(frames=(None, None), gate=gate))
+
+    leader = asyncio.create_task(capture_camera_frame_bytes("10.0.2.43", "code", "P2S"))
+    await _let_leader_start(capture)
+    first = asyncio.create_task(capture_camera_frame_bytes("10.0.2.43", "code", "P2S"))
+    await asyncio.sleep(0)
+    second = asyncio.create_task(capture_camera_frame_bytes("10.0.2.43", "code", "P2S"))
+    await asyncio.sleep(0)
+    gate.set()
+
+    assert await leader is None
+    assert await first is None
+    assert await second is None
+    # The leader's capture plus one retry — not one per disappointed caller.
+    assert capture.count == 2
+
+
+@pytest.mark.asyncio
+async def test_follower_timeout_does_not_sabotage_the_capture(patch_capture):
+    """A follower giving up leaves the capture running for everyone else.
+
+    The call sites disagree about the timeout (10s plate detection, 20s Obico),
+    so a follower must be able to abandon a join without cancelling a capture
+    other callers are still waiting on.
+    """
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(gate=gate))
+
+    leader = asyncio.create_task(capture_camera_frame_bytes("10.0.2.43", "code", "P2S", timeout=30))
+    await _let_leader_start(capture)
+    impatient = asyncio.create_task(capture_camera_frame_bytes("10.0.2.43", "code", "P2S", timeout=0.01))
+    patient = asyncio.create_task(capture_camera_frame_bytes("10.0.2.43", "code", "P2S", timeout=30))
+
+    assert await impatient is None  # gave up on its own deadline
+    gate.set()
+
+    assert await leader == FRAME_A
+    assert await patient == FRAME_A  # unaffected by the one that walked away
+    assert capture.count == 1
+
+
+@pytest.mark.asyncio
+async def test_cancelled_leader_still_delivers_to_followers(patch_capture):
+    """Snapshot requests get cancelled routinely (client navigates away).
+
+    The follower must not lose the frame because the caller that happened to
+    open the connection went away.
+    """
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(gate=gate))
+
+    leader = asyncio.create_task(capture_camera_frame_bytes("10.0.2.43", "code", "P2S"))
+    await _let_leader_start(capture)
+    follower = asyncio.create_task(capture_camera_frame_bytes("10.0.2.43", "code", "P2S"))
+    await asyncio.sleep(0)
+
+    leader.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await leader
+    gate.set()
+
+    assert await follower == FRAME_A
+    assert capture.count == 1
+
+
+@pytest.mark.asyncio
+async def test_cancelling_a_follower_leaves_the_leader_alone(patch_capture):
+    """The mirror case: the follower's cancellation is its own business."""
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(gate=gate))
+
+    leader = asyncio.create_task(capture_camera_frame_bytes("10.0.2.43", "code", "P2S"))
+    await _let_leader_start(capture)
+    follower = asyncio.create_task(capture_camera_frame_bytes("10.0.2.43", "code", "P2S"))
+    await asyncio.sleep(0)
+
+    follower.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await follower
+    gate.set()
+
+    assert await leader == FRAME_A
+    assert capture.count == 1
+
+
+@pytest.mark.asyncio
+async def test_capture_in_flight_reports_the_window(patch_capture):
+    """The predicate the diagnose tool uses to know it will join, not measure."""
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(gate=gate))
+
+    assert capture_in_flight("10.0.2.43") is False
+
+    leader = asyncio.create_task(capture_camera_frame_bytes("10.0.2.43", "code", "P2S"))
+    await _let_leader_start(capture)
+
+    assert capture_in_flight("10.0.2.43") is True
+    assert capture_in_flight("10.0.2.44") is False  # per printer
+
+    gate.set()
+    await leader
+    await asyncio.sleep(0)
+
+    assert capture_in_flight("10.0.2.43") is False

backend/tests/unit/services/test_camera_diagnose.py

@@ -221,6 +221,79 @@ class TestFirstFrameStage:
         assert result.overall_status == "ok"
         assert result.summary_code == "all_ok"
         assert all(s.status == "ok" for s in result.stages)
+        assert result.stages[1].code is None  # we opened our own connection
+
+    @pytest.mark.asyncio
+    async def test_pass_riding_on_an_inflight_capture_says_so(self):
+        """#2705: a capture already running means we get its frame, not our own.
+
+        The frame is real evidence the camera works, so the stage still passes —
+        but duration_ms is then mostly time spent queueing behind someone
+        else's capture, and claiming a connection we never opened is exactly
+        what a diagnostic must not do."""
+
+        async def _tcp_ok(*_a, **_kw):
+            writer = AsyncMock()
+            return AsyncMock(), writer
+
+        with (
+            patch(
+                "backend.app.services.camera_diagnose.asyncio.open_connection",
+                new=_tcp_ok,
+            ),
+            patch(
+                "backend.app.services.camera_diagnose.capture_in_flight",
+                return_value=True,
+            ),
+            patch(
+                "backend.app.services.camera_diagnose.capture_camera_frame_bytes",
+                new_callable=AsyncMock,
+                return_value=b"\xff\xd8\xff\xd9",
+            ),
+        ):
+            result = await diagnose_camera(
+                ip_address="192.0.2.1",
+                access_code="x",
+                model="P2S",
+                printer_id=1,
+            )
+        assert result.overall_status == "ok"
+        assert result.summary_code == "all_ok"
+        assert result.stages[1].status == "ok"
+        assert result.stages[1].code == "coalesced_capture"
+
+    @pytest.mark.asyncio
+    async def test_failure_is_not_annotated_as_coalesced(self):
+        """A follower whose leader fails goes on to capture on its own, so a
+        failure here was this stage's own attempt — no qualifier needed."""
+
+        async def _tcp_ok(*_a, **_kw):
+            writer = AsyncMock()
+            return AsyncMock(), writer
+
+        with (
+            patch(
+                "backend.app.services.camera_diagnose.asyncio.open_connection",
+                new=_tcp_ok,
+            ),
+            patch(
+                "backend.app.services.camera_diagnose.capture_in_flight",
+                return_value=True,
+            ),
+            patch(
+                "backend.app.services.camera_diagnose.capture_camera_frame_bytes",
+                new_callable=AsyncMock,
+                return_value=None,
+            ),
+        ):
+            result = await diagnose_camera(
+                ip_address="192.0.2.1",
+                access_code="x",
+                model="P2S",
+                printer_id=1,
+            )
+        assert result.summary_code == "no_frame"
+        assert result.stages[1].code == "no_frame"
 
 
 class TestResultMetadata: