fix(camera): rotate every still exactly once, and cover the sources that let ffmpeg write the file

Review follow-ups on applying camera_rotation to finish photos and
layer-timelapse frames.

Rotating the frame popped from _stage22_finish_frames rotated one of its
sources twice. The cache has two kinds of feeder: live grabs, which are raw,
and the #1867 in-print bank, whose bytes come from
_capture_snapshot_for_notification and have already been rotated on the way
in. The consumer cannot tell them apart, so on the finish_state trigger - the
path the bank exists to serve, on firmware that never emits stg_cur=22 - a 180
degree rotation cancelled itself out and the photo was upside-down again,
which is the reported symptom exactly; 90 and 270 landed 180 out. Rotation now
happens where each frame is captured, so every entry in the cache carries one
rotation whatever produced it, and the invariant is stated both where the
cache is declared and where it is consumed.

Two finish-photo sources were still writing unrotated files: the built-in
camera's own capture_finish_photo, and the still extracted from a
printer-recorded timelapse - which is the *preferred* source for a built-in
camera print, so a user with a rotation set got a correctly oriented photo or
not depending on which source happened to win. Neither ever holds the frame as
bytes; ffmpeg writes the file and they return a filename. apply_camera_rotation_to_file
handles that case and is best-effort - a failed rotate leaves the unrotated
file rather than losing a delivered photo. The archived video itself is the
printer's own file and is not re-encoded, so it still plays at the camera's
native orientation; the CHANGELOG says so rather than leaving it to be
discovered.

apply_camera_rotation logs at debug, not info. It was on a path that runs once
per layer, where a tall print would have put hundreds of lines in the log for
something the surrounding capture already reports at debug.

The moved rotation logic had no test of its own - every existing test patches
it out and asserts the call, so a flipped sign or a dropped expand=True would
have shipped green. test_camera_rotation.py drives the real round trip: a
corner marker pins which way it turns, the dimensions pin that the frame is
not cropped, and an undecodable frame comes back by identity because a capture
path must not lose a frame to a failed rotate.

Tests for the fix itself sit on both sides of the cache. The producer half is
driven directly; the consumer half is a closure nested inside on_print_complete
with nothing able to reach it, so it is pinned by an AST guard - checked
against the source because the alternative is no check at all. Reverting
main.py to the pre-fix shape fails three of the five, the guard among them.

The three new tests used Path("/tmp/test") for a patched base_dir, which Bandit
flagged (B108); they take tmp_path now.

backend/app/main.py

@@ -347,6 +347,12 @@ _active_prints: dict[tuple[int, str], int] = {}
 # captures the better-framed pre-bed-drop moment without us having to force
 # timelapse on at dispatch (the #1397 mechanism that caused #1721's per-layer
 # nozzle parking on slicer profiles with Timelapse Type = Smooth).
+#
+# #2708: the bytes in here are ALWAYS already rotated by the printer's
+# camera_rotation. `on_finish_photo_moment` owns that, because one of its
+# sources (the #1867 in-print bank) is rotated before it ever reaches the
+# bank and the others are not — so the consumer can't tell them apart and
+# must not rotate again.
 _stage22_finish_frames: dict[int, bytes] = {}
 
 # #1790: per-printer producer-done event. Set by `on_finish_photo_moment` in its
@@ -3982,6 +3988,7 @@ async def _capture_finish_photo_from_timelapse(
     archive_id: int,
     archive_dir: Path,
     timeout: float | None = None,
+    rotation: int = 0,
 ) -> tuple[str | None, bool]:
     """Wait for the per-print timelapse to land on the archive and extract its
     last frame as the finish photo (#1397).
@@ -4001,11 +4008,16 @@ async def _capture_finish_photo_from_timelapse(
     video landed (whether or not extraction worked), because in that case
     waiting longer changes nothing. The caller uses that to decide between
     falling back permanently and scheduling a background upgrade.
+
+    ``rotation`` is the printer's camera_rotation, applied to the extracted
+    still (#2708) so this source agrees with every other finish-photo source.
+    The archived video itself is the printer's own file and is left alone —
+    rotating it would mean re-encoding it.
     """
     import uuid
 
     from backend.app.models.archive import PrintArchive
-    from backend.app.services.camera import extract_video_last_frame
+    from backend.app.services.camera import apply_camera_rotation_to_file, extract_video_last_frame
 
     logger = logging.getLogger(__name__)
 
@@ -4028,6 +4040,7 @@ async def _capture_finish_photo_from_timelapse(
                 filename = f"finish_{timestamp}_{uuid.uuid4().hex[:8]}.jpg"
                 output_path = photos_dir / filename
                 if await extract_video_last_frame(video_path, output_path):
+                    await apply_camera_rotation_to_file(output_path, rotation, logger)
                     logger.info(
                         "[PHOTO-BG] Extracted finish photo from timelapse %s for archive %s",
                         video_path.name,
@@ -4052,7 +4065,7 @@ async def _capture_finish_photo_from_timelapse(
         await asyncio.sleep(poll_interval)
 
 
-async def _upgrade_finish_photo_from_timelapse(archive_id: int, archive_dir: Path) -> None:
+async def _upgrade_finish_photo_from_timelapse(archive_id: int, archive_dir: Path, rotation: int = 0) -> None:
     """Add the timelapse's last frame to an archive after the fact (#2704).
 
     The print-complete notification waits only ~60s for the video, because
@@ -4071,7 +4084,7 @@ async def _upgrade_finish_photo_from_timelapse(archive_id: int, archive_dir: Pat
     logger = logging.getLogger(__name__)
 
     filename, _ = await _capture_finish_photo_from_timelapse(
-        archive_id, archive_dir, timeout=_FINISH_PHOTO_UPGRADE_TIMEOUT_SECONDS
+        archive_id, archive_dir, timeout=_FINISH_PHOTO_UPGRADE_TIMEOUT_SECONDS, rotation=rotation
     )
     if not filename:
         logger.info("[PHOTO-UPGRADE] No timelapse frame for archive %s; keeping the live grab", archive_id)
@@ -4356,6 +4369,11 @@ async def on_finish_photo_moment(printer_id: int, data: dict):
                 return
 
         frame_bytes: bytes | None = None
+        # #2708: the banked frame arrives already rotated — it comes from
+        # `_capture_snapshot_for_notification`, which rotates before returning.
+        # Every other source below is a raw grab. Tracking which lets us store
+        # exactly one rotation in `_stage22_finish_frames` either way.
+        frame_already_rotated = False
 
         # #1867: on the FINISH-state fallback the End G-code (e.g. SwapMod
         # plate-swap) has already run, so a live grab now captures the swapped
@@ -4367,6 +4385,7 @@ async def on_finish_photo_moment(printer_id: int, data: dict):
             banked = _inprint_frame_bank.get(printer_id)
             if banked:
                 frame_bytes = banked
+                frame_already_rotated = True
                 logger.info(
                     "[FINISH-PHOTO-MOMENT] using banked in-print frame (%d bytes) — "
                     "avoids post-swap live grab on stage-22-less firmware",
@@ -4420,6 +4439,8 @@ async def on_finish_photo_moment(printer_id: int, data: dict):
                     )
 
         if frame_bytes:
+            if not frame_already_rotated:
+                frame_bytes = _apply_camera_rotation(frame_bytes, printer, logger)
             _stage22_finish_frames[printer_id] = frame_bytes
         else:
             logger.warning(
@@ -5307,6 +5328,7 @@ async def on_print_complete(printer_id: int, data: dict):
                 photo_filename, timelapse_still_pending = await _capture_finish_photo_from_timelapse(
                     archive_id=archive_id,
                     archive_dir=archive_dir,
+                    rotation=getattr(printer, "camera_rotation", 0),
                 )
 
             # #1721: replacement framing path — on_finish_photo_moment
@@ -5334,7 +5356,9 @@ async def on_print_complete(printer_id: int, data: dict):
                         )
                 cached_frame = _stage22_finish_frames.pop(printer_id, None)
                 if cached_frame:
-                    cached_frame = _apply_camera_rotation(cached_frame, printer, logger)
+                    # Already rotated by the producer (#2708) — rotating again
+                    # here would undo the fix on the banked-frame path, whose
+                    # bytes reach the cache having been rotated once already.
                     photos_dir = archive_dir / "photos"
                     photos_dir.mkdir(parents=True, exist_ok=True)
                     timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
@@ -5405,6 +5429,7 @@ async def on_print_complete(printer_id: int, data: dict):
                             access_code=printer.access_code,
                             model=printer.model,
                             archive_dir=archive_dir,
+                            rotation=getattr(printer, "camera_rotation", 0),
                         )
 
             # Write phase: attach the photo in a fresh short-lived session.
@@ -5436,7 +5461,9 @@ async def on_print_complete(printer_id: int, data: dict):
             # gallery never lists.
             if timelapse_still_pending:
                 spawn_background_task(
-                    _upgrade_finish_photo_from_timelapse(archive_id, archive_dir),
+                    _upgrade_finish_photo_from_timelapse(
+                        archive_id, archive_dir, rotation=getattr(printer, "camera_rotation", 0)
+                    ),
                     name=f"finish-photo-upgrade-{archive_id}",
                 )
 

backend/app/services/camera.py

@@ -843,6 +843,10 @@ def apply_camera_rotation(image_data: bytes, rotation: int, logger: logging.Logg
     snapshots, finish photos, layer-timelapse frames) - previously only
     wired into the notification-snapshot path, which left finish photos
     and timelapse videos upside-down whenever camera_rotation was set.
+
+    Returns *image_data* itself (identity, not a copy) when there is nothing
+    to do or the rotate fails; callers that write to disk use that to skip a
+    pointless rewrite.
     """
     if not rotation:
         return image_data
@@ -858,19 +862,46 @@ def apply_camera_rotation(image_data: bytes, rotation: int, logger: logging.Logg
         buf = BytesIO()
         img.save(buf, format="JPEG", quality=90)
         rotated = buf.getvalue()
-        logger.info("Applied %d° camera rotation: %s → %s bytes", rotation, len(image_data), len(rotated))
+        # Debug, not info: layer-timelapse calls this once per layer, so a tall
+        # print would otherwise put hundreds of lines in the log for something
+        # the surrounding capture already reports at debug level.
+        logger.debug("Applied %d° camera rotation: %s → %s bytes", rotation, len(image_data), len(rotated))
         return rotated
     except Exception as e:
         logger.warning("Failed to apply camera rotation: %s", e)
         return image_data
 
 
+async def apply_camera_rotation_to_file(path: Path, rotation: int, logger: logging.Logger) -> None:
+    """Rotate a JPEG that has already been written to disk, in place.
+
+    Two finish-photo sources never hold the frame as bytes - ``ffmpeg`` writes
+    the file for them, and they return only a filename - so they can't use
+    ``apply_camera_rotation`` directly. Best-effort: any failure leaves the
+    unrotated file in place, which is what the caller had before.
+    """
+    if not rotation:
+        return
+
+    try:
+        data = await asyncio.to_thread(path.read_bytes)
+        rotated = await asyncio.to_thread(apply_camera_rotation, data, rotation, logger)
+        if rotated is data:
+            # Nothing was done (the rotate failed and returned its input) -
+            # rewriting the same bytes would only risk truncating a good file.
+            return
+        await asyncio.to_thread(path.write_bytes, rotated)
+    except Exception as e:
+        logger.warning("Failed to rotate %s in place: %s", path.name, e)
+
+
 async def capture_finish_photo(
     printer_id: int,
     ip_address: str,
     access_code: str,
     model: str | None,
     archive_dir: Path,
+    rotation: int = 0,
 ) -> str | None:
     """Capture a finish photo and save it to the archive's photos folder.
 
@@ -880,6 +911,9 @@ async def capture_finish_photo(
         access_code: Printer access code
         model: Printer model
         archive_dir: Directory of the archive (where the 3MF is stored)
+        rotation: Printer's configured camera_rotation (degrees clockwise).
+            ffmpeg writes the file directly here, so the rotation is applied
+            to it afterwards rather than to bytes in hand.
 
     Returns:
         Filename of the captured photo, or None if capture failed
@@ -904,6 +938,7 @@ async def capture_finish_photo(
     )
 
     if success:
+        await apply_camera_rotation_to_file(output_path, rotation, logger)
         logger.info("Finish photo saved: %s", filename)
         return filename
     else:

backend/tests/unit/services/test_camera_rotation.py

@@ -0,0 +1,146 @@
+"""Tests for the shared camera-rotation helpers (#2708).
+
+Every other test of a rotating path patches ``apply_camera_rotation`` out and
+asserts the call, which proves the wiring but not the rotation. These drive
+the real PIL round trip, so a flipped sign or a dropped ``expand=True`` fails
+here rather than shipping.
+"""
+
+import io
+import logging
+
+import pytest
+from PIL import Image
+
+from backend.app.services.camera import apply_camera_rotation, apply_camera_rotation_to_file
+
+logger = logging.getLogger(__name__)
+
+
+def _jpeg(width: int, height: int, corner: tuple[int, int, int] = (255, 0, 0)) -> bytes:
+    """A JPEG with one distinctly coloured pixel block in the top-left corner,
+    so which way it turned is observable and not just the dimensions."""
+    img = Image.new("RGB", (width, height), (0, 0, 255))
+    for x in range(min(8, width)):
+        for y in range(min(8, height)):
+            img.putpixel((x, y), corner)
+    buf = io.BytesIO()
+    img.save(buf, format="JPEG", quality=95)
+    return buf.getvalue()
+
+
+def _open(data: bytes) -> Image.Image:
+    return Image.open(io.BytesIO(data))
+
+
+def _brightest_corner(img: Image.Image) -> str:
+    """Which corner holds the red block, sampled a few pixels in to stay clear
+    of JPEG ringing at the edges."""
+    w, h = img.size
+    probes = {
+        "top-left": (3, 3),
+        "top-right": (w - 4, 3),
+        "bottom-left": (3, h - 4),
+        "bottom-right": (w - 4, h - 4),
+    }
+    return max(probes, key=lambda name: img.getpixel(probes[name])[0] - img.getpixel(probes[name])[2])
+
+
+class TestApplyCameraRotation:
+    def test_zero_rotation_returns_the_input_object(self):
+        """Not merely equal — identity. apply_camera_rotation_to_file uses this
+        to decide there is nothing to write back."""
+        src = _jpeg(64, 32)
+        assert apply_camera_rotation(src, 0, logger) is src
+
+    def test_90_degrees_turns_clockwise(self):
+        """camera_rotation is documented as degrees *clockwise*, and PIL's
+        rotate() is counter-clockwise — the helper negates to compensate. A
+        lost negation would send the corner to bottom-right instead."""
+        src = _jpeg(64, 32)
+        assert _brightest_corner(_open(src)) == "top-left"
+
+        out = _open(apply_camera_rotation(src, 90, logger))
+        assert out.size == (32, 64)  # expand=True, so the frame is not cropped
+        assert _brightest_corner(out) == "top-right"
+
+    def test_270_degrees_turns_the_other_way(self):
+        out = _open(apply_camera_rotation(_jpeg(64, 32), 270, logger))
+        assert out.size == (32, 64)
+        assert _brightest_corner(out) == "bottom-left"
+
+    def test_180_degrees_keeps_the_dimensions_and_flips_the_corner(self):
+        out = _open(apply_camera_rotation(_jpeg(64, 32), 180, logger))
+        assert out.size == (64, 32)
+        assert _brightest_corner(out) == "bottom-right"
+
+    def test_applying_180_twice_is_the_bug_that_was_fixed(self):
+        """The regression this guards: two rotations cancel out and the photo
+        is upside-down again. Kept as a test so the invariant that
+        _stage22_finish_frames holds exactly one rotation has a stated reason.
+        """
+        src = _jpeg(64, 32)
+        once = apply_camera_rotation(src, 180, logger)
+        twice = apply_camera_rotation(once, 180, logger)
+        assert _brightest_corner(_open(once)) == "bottom-right"
+        assert _brightest_corner(_open(twice)) == "top-left"  # back to the original
+
+    def test_undecodable_bytes_return_unchanged(self):
+        """A capture path must not lose a frame because the rotate failed —
+        an unrotated photo beats no photo."""
+        junk = b"not a jpeg at all"
+        assert apply_camera_rotation(junk, 90, logger) is junk
+
+    def test_a_failed_rotate_is_logged_as_a_warning(self, caplog):
+        with caplog.at_level(logging.WARNING, logger=__name__):
+            apply_camera_rotation(b"not a jpeg at all", 90, logger)
+        assert any("Failed to apply camera rotation" in r.message for r in caplog.records)
+
+    def test_a_successful_rotate_does_not_log_at_info(self, caplog):
+        """Layer-timelapse calls this once per layer; at INFO a tall print
+        would bury the log."""
+        with caplog.at_level(logging.INFO, logger=__name__):
+            apply_camera_rotation(_jpeg(64, 32), 90, logger)
+        assert caplog.records == []
+
+
+class TestApplyCameraRotationToFile:
+    """The two finish-photo sources that let ffmpeg write the file and never
+    hold the bytes: capture_finish_photo and the timelapse last-frame extract."""
+
+    @pytest.mark.asyncio
+    async def test_rotates_in_place(self, tmp_path):
+        path = tmp_path / "finish.jpg"
+        path.write_bytes(_jpeg(64, 32))
+
+        await apply_camera_rotation_to_file(path, 90, logger)
+
+        out = _open(path.read_bytes())
+        assert out.size == (32, 64)
+        assert _brightest_corner(out) == "top-right"
+
+    @pytest.mark.asyncio
+    async def test_zero_rotation_leaves_the_file_untouched(self, tmp_path):
+        path = tmp_path / "finish.jpg"
+        original = _jpeg(64, 32)
+        path.write_bytes(original)
+
+        await apply_camera_rotation_to_file(path, 0, logger)
+
+        assert path.read_bytes() == original
+
+    @pytest.mark.asyncio
+    async def test_a_file_that_cannot_be_rotated_is_left_intact(self, tmp_path):
+        """Not truncated, not deleted — the caller's unrotated photo survives."""
+        path = tmp_path / "finish.jpg"
+        path.write_bytes(b"not a jpeg at all")
+
+        await apply_camera_rotation_to_file(path, 90, logger)
+
+        assert path.read_bytes() == b"not a jpeg at all"
+
+    @pytest.mark.asyncio
+    async def test_a_missing_file_does_not_raise(self, tmp_path):
+        """Best-effort: this runs after the capture reported success, and must
+        not turn a delivered photo into a failed one."""
+        await apply_camera_rotation_to_file(tmp_path / "gone.jpg", 90, logger)

backend/tests/unit/services/test_layer_timelapse.py

@@ -254,11 +254,11 @@ class TestCaptureLayerAppliesRotation:
     buffer, before writing to disk."""
 
     @pytest.mark.asyncio
-    async def test_rotates_fresh_capture_when_configured(self):
+    async def test_rotates_fresh_capture_when_configured(self, tmp_path):
         from backend.app.services.layer_timelapse import TimelapseSession
 
         with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
-            mock_settings.base_dir = Path("/tmp/test")
+            mock_settings.base_dir = tmp_path
 
             with patch.object(Path, "mkdir"):
                 session = TimelapseSession(1, 100, "/dev/video1", "usb", rotation=180)
@@ -283,11 +283,11 @@ class TestCaptureLayerAppliesRotation:
         mock_write.assert_called_once_with(b"\xff\xd8rotated\xff\xd9")
 
     @pytest.mark.asyncio
-    async def test_rotates_buffered_frame_when_configured(self):
+    async def test_rotates_buffered_frame_when_configured(self, tmp_path):
         from backend.app.services.layer_timelapse import TimelapseSession
 
         with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
-            mock_settings.base_dir = Path("/tmp/test")
+            mock_settings.base_dir = tmp_path
 
             with patch.object(Path, "mkdir"):
                 session = TimelapseSession(1, 100, "/dev/video1", "usb", rotation=90)
@@ -310,13 +310,13 @@ class TestCaptureLayerAppliesRotation:
         mock_write.assert_called_once_with(b"\xff\xd8rotated\xff\xd9")
 
     @pytest.mark.asyncio
-    async def test_skips_rotation_when_not_configured(self):
+    async def test_skips_rotation_when_not_configured(self, tmp_path):
         """Default rotation=0 - no-op, and must not even call apply_camera_rotation
         (avoids the PIL decode/re-encode round trip for the common case)."""
         from backend.app.services.layer_timelapse import TimelapseSession
 
         with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
-            mock_settings.base_dir = Path("/tmp/test")
+            mock_settings.base_dir = tmp_path
 
             with patch.object(Path, "mkdir"):
                 session = TimelapseSession(1, 100, "/dev/video1", "usb")

backend/tests/unit/test_finish_photo_from_timelapse.py

@@ -179,3 +179,69 @@ async def test_polls_until_file_appears(tmp_path: Path, patched_session, monkeyp
 
     assert result is not None
     assert result.startswith("finish_")
+
+
+async def test_extracted_frame_is_rotated_when_configured(tmp_path: Path, patched_session, monkeypatch):
+    """#2708: this source hands a path to ffmpeg and never holds the bytes, so
+    it was the one finish-photo source that ignored camera_rotation entirely.
+    A built-in-camera print with a timelapse prefers this source over the live
+    grab, so leaving it out meant the orientation depended on which source won.
+    """
+    import io
+
+    from PIL import Image
+
+    monkeypatch.setattr(main_module.app_settings, "base_dir", tmp_path)
+    video_relpath = Path("archive/1/print/timelapse.mp4")
+    video_abspath = tmp_path / video_relpath
+    video_abspath.parent.mkdir(parents=True, exist_ok=True)
+    video_abspath.write_bytes(b"x" * 100)
+
+    async def fake_extract(src, dst):
+        buf = io.BytesIO()
+        Image.new("RGB", (64, 32), (0, 0, 255)).save(buf, format="JPEG")
+        dst.write_bytes(buf.getvalue())
+        return True
+
+    monkeypatch.setattr(main_module, "_FINISH_PHOTO_TIMELAPSE_POLL_INTERVAL_SECONDS", 0.0)
+    patched_session.timelapse_path = str(video_relpath)
+
+    with patch("backend.app.services.camera.extract_video_last_frame", new=fake_extract):
+        result, _ = await _capture_finish_photo_from_timelapse(
+            archive_id=42,
+            archive_dir=tmp_path / "archive_dir",
+            rotation=90,
+        )
+
+    assert result is not None
+    written = tmp_path / "archive_dir" / "photos" / result
+    # 64x32 turned a quarter turn: the file on disk is the rotated one, not
+    # what ffmpeg wrote.
+    assert Image.open(io.BytesIO(written.read_bytes())).size == (32, 64)
+
+
+async def test_extracted_frame_is_untouched_without_a_rotation(tmp_path: Path, patched_session, monkeypatch):
+    """The default path must not decode and re-encode ffmpeg's output for
+    nothing — that would cost a generation of JPEG quality on every print."""
+    monkeypatch.setattr(main_module.app_settings, "base_dir", tmp_path)
+    video_relpath = Path("archive/1/print/timelapse.mp4")
+    video_abspath = tmp_path / video_relpath
+    video_abspath.parent.mkdir(parents=True, exist_ok=True)
+    video_abspath.write_bytes(b"x" * 100)
+
+    extracted = b"\xff\xd8" + b"\x00" * 50
+
+    async def fake_extract(src, dst):
+        dst.write_bytes(extracted)
+        return True
+
+    monkeypatch.setattr(main_module, "_FINISH_PHOTO_TIMELAPSE_POLL_INTERVAL_SECONDS", 0.0)
+    patched_session.timelapse_path = str(video_relpath)
+
+    with patch("backend.app.services.camera.extract_video_last_frame", new=fake_extract):
+        result, _ = await _capture_finish_photo_from_timelapse(
+            archive_id=42,
+            archive_dir=tmp_path / "archive_dir",
+        )
+
+    assert (tmp_path / "archive_dir" / "photos" / result).read_bytes() == extracted

backend/tests/unit/test_finish_photo_moment_sync.py

@@ -322,3 +322,139 @@ async def test_bank_skips_during_calibration_substage(monkeypatch):
     _bank_env(monkeypatch, sub_stage=14)
     await main_module._maybe_bank_inprint_frame(3, 2)
     assert 3 not in main_module._inprint_frame_bank
+
+
+class TestStage22CacheHoldsExactlyOneRotation:
+    """#2708. `_stage22_finish_frames` is fed from two kinds of source: live
+    grabs, which are raw, and the #1867 in-print bank, whose bytes came from
+    `_capture_snapshot_for_notification` and are therefore ALREADY rotated.
+    The consumer cannot tell them apart, so the producer normalises: every
+    entry in the cache has had the rotation applied exactly once.
+
+    Rotating on the consumer side instead put two rotations on the banked
+    path — at 180 degrees that is the reported bug reproduced exactly, and at
+    90/270 it lands the photo 180 degrees out.
+    """
+
+    @staticmethod
+    def _jpeg(width, height):
+        import io
+
+        from PIL import Image
+
+        buf = io.BytesIO()
+        Image.new("RGB", (width, height), (0, 0, 255)).save(buf, format="JPEG")
+        return buf.getvalue()
+
+    @staticmethod
+    def _size(data):
+        import io
+
+        from PIL import Image
+
+        return Image.open(io.BytesIO(data)).size
+
+    async def test_a_live_grab_is_rotated_before_caching(self, patched_env, monkeypatch):
+        monkeypatch.setattr(patched_env, "camera_rotation", 90, raising=False)
+        raw = self._jpeg(64, 32)
+
+        async def _capture(**_kwargs):
+            return raw
+
+        monkeypatch.setattr("backend.app.services.camera.capture_camera_frame_bytes", _capture)
+
+        await on_finish_photo_moment(patched_env.id, {"trigger": "finish_state"})
+
+        cached = main_module._stage22_finish_frames[patched_env.id]
+        assert self._size(cached) == (32, 64)
+
+    async def test_the_banked_frame_is_cached_verbatim(self, patched_env, monkeypatch):
+        """The bank is filled by `_capture_snapshot_for_notification`, which
+        rotates before it returns — so the producer must pass those bytes
+        through untouched rather than rotating them a second time.
+
+        Note this pins the invariant forward; it does not on its own prove the
+        bug fixed, because the old producer didn't rotate anything either. The
+        pair that discriminates is `test_a_live_grab_is_rotated_before_caching`
+        (producer now rotates) plus the source guard below (consumer no longer
+        does).
+        """
+        monkeypatch.setattr(patched_env, "camera_rotation", 90, raising=False)
+        already_rotated = self._jpeg(32, 64)  # what one rotation of a 64x32 frame looks like
+        main_module._inprint_frame_bank[patched_env.id] = already_rotated
+
+        async def _capture(**_kwargs):  # pragma: no cover - must not be reached
+            raise AssertionError("the banked frame should have been preferred")
+
+        monkeypatch.setattr("backend.app.services.camera.capture_camera_frame_bytes", _capture)
+
+        await on_finish_photo_moment(patched_env.id, {"trigger": "finish_state"})
+
+        cached = main_module._stage22_finish_frames[patched_env.id]
+        assert cached is already_rotated
+        assert self._size(cached) == (32, 64)
+
+    async def test_a_stage22_grab_is_rotated_even_though_the_bank_is_full(self, patched_env, monkeypatch):
+        """Only the `finish_state` trigger reads the bank. The `stage_22` and
+        `last_layer` triggers take a live grab, which still needs rotating —
+        a shared "did we use the bank" flag must not latch on the bank merely
+        existing."""
+        monkeypatch.setattr(patched_env, "camera_rotation", 90, raising=False)
+        main_module._inprint_frame_bank[patched_env.id] = self._jpeg(999, 1)
+
+        async def _capture(**_kwargs):
+            return self._jpeg(64, 32)
+
+        monkeypatch.setattr("backend.app.services.camera.capture_camera_frame_bytes", _capture)
+
+        await on_finish_photo_moment(patched_env.id, {"trigger": "stage_22"})
+
+        cached = main_module._stage22_finish_frames[patched_env.id]
+        assert self._size(cached) == (32, 64)
+
+    async def test_no_rotation_configured_caches_the_bytes_as_captured(self, patched_env, monkeypatch):
+        raw = self._jpeg(64, 32)
+
+        async def _capture(**_kwargs):
+            return raw
+
+        monkeypatch.setattr("backend.app.services.camera.capture_camera_frame_bytes", _capture)
+
+        await on_finish_photo_moment(patched_env.id, {"trigger": "finish_state"})
+
+        assert main_module._stage22_finish_frames[patched_env.id] is raw
+
+
+def test_the_consumer_does_not_rotate_the_cached_frame():
+    """The other half of the #2708 invariant, and the half with no runtime
+    harness: `_background_finish_photo` is a closure nested inside
+    `on_print_complete`, so nothing can drive its cached-frame branch
+    directly. What it must NOT do is rotate what it pops from
+    `_stage22_finish_frames` — the producer has already done that, and doing
+    it again upside-downs the banked path, which is the bug this fixed.
+
+    Checked against the source because the alternative is no check at all.
+    """
+    import ast
+    from pathlib import Path
+
+    main_py = Path(__file__).resolve().parents[2] / "app" / "main.py"
+    assert main_py.exists(), f"guard is looking in the wrong place: {main_py}"
+    tree = ast.parse(main_py.read_text())
+
+    offenders = [
+        node.lineno
+        for node in ast.walk(tree)
+        if isinstance(node, ast.Call)
+        and isinstance(node.func, ast.Name)
+        and node.func.id == "_apply_camera_rotation"
+        and node.args
+        and isinstance(node.args[0], ast.Name)
+        and node.args[0].id == "cached_frame"
+    ]
+
+    assert not offenders, (
+        f"main.py:{offenders} rotates the frame popped from _stage22_finish_frames. "
+        "Those bytes are already rotated by on_finish_photo_moment (#2708); rotating "
+        "again returns a 180-degree print to upside-down."
+    )