diff --git a/client/pyroclient/client.py b/client/pyroclient/client.py
index 9f617bb1..3f07926e 100644
--- a/client/pyroclient/client.py
+++ b/client/pyroclient/client.py
@@ -496,9 +496,11 @@ def fetch_latest_sequences(self) -> Response:
     def fetch_sequences_detections(
         self,
         sequence_id: int,
-        limit: int = 10,
+        limit: Union[int, None] = None,
         desc: bool = True,
         with_crop: bool = True,
+        sampling: int = 1,
+        offset: int = 0,
     ) -> Response:
         """List the detections of a sequence
 
@@ -508,17 +510,33 @@ def fetch_sequences_detections(
 
         Args:
             sequence_id: ID of the associated sequence entry
-            limit: maximum number of detections to fetch
+            limit: maximum number of detections to fetch. Left unset (the default) the API picks
+                10, or with sampling the size of the whole sampled span capped at 500
             desc: whether to order the detections by created_at in descending order
             with_crop: whether to include the crop_url for detections that have a crop
+            sampling: keep one detection every N (1 = all, max 10000); the kept frames are picked
+                chronologically and do not depend on desc. Pair it with limit unset to span the
+                sequence, since an explicit limit below ceil(detections_count / sampling) returns
+                only part of the span. The X-Sampled-Total and X-Sampled-Truncated response
+                headers say whether what came back covers the sequence
+            offset: number of detections to skip, within the sampled set
 
         Returns:
             HTTP response
         """
+        params: Dict[str, Any] = {
+            "desc": desc,
+            "with_crop": with_crop,
+            "sampling": sampling,
+            "offset": offset,
+        }
+        # Omitted rather than defaulted client-side, so the API can size it from the sampled set.
+        if limit is not None:
+            params["limit"] = limit
         return requests.get(
             urljoin(self._route_prefix, ClientRoute.SEQUENCES_FETCH_DETECTIONS.format(seq_id=sequence_id)),
             headers=self.headers,
-            params={"limit": limit, "desc": desc, "with_crop": with_crop},
+            params=params,
             timeout=self.timeout,
         )
 
diff --git a/client/tests/test_client.py b/client/tests/test_client.py
index d98e620e..35425987 100644
--- a/client/tests/test_client.py
+++ b/client/tests/test_client.py
@@ -191,9 +191,20 @@ def test_user_workflow(test_cam_workflow, user_token):
     assert len(response.json()) == 0  # Sequence was labeled by agent
     response = user_client.fetch_sequences_from_date(datetime.utcnow().date().isoformat())
     assert len(response.json()) == 1
-    response = user_client.fetch_sequences_detections(response.json()[0]["id"])
+    sequence_id = response.json()[0]["id"]
+    response = user_client.fetch_sequences_detections(sequence_id)
     assert response.status_code == 200, response.__dict__
     # 4 real detections + the continuity row added by the empty frame in test_cam_workflow
     detections = response.json()
     assert len(detections) == 5
     assert sum(det["bbox"] == "[]" for det in detections) == 1
+    # An explicit limit is forwarded, an omitted one is left to the API.
+    response = user_client.fetch_sequences_detections(sequence_id, limit=2)
+    assert response.status_code == 200, response.__dict__
+    assert len(response.json()) == 2
+    # With sampling and no limit, the API sizes the response to span the sampled set.
+    response = user_client.fetch_sequences_detections(sequence_id, sampling=2)
+    assert response.status_code == 200, response.__dict__
+    assert len(response.json()) == 3  # ceil(5 / 2)
+    assert response.headers["x-sampled-total"] == "3"
+    assert response.headers["x-sampled-truncated"] == "false"
diff --git a/src/app/api/api_v1/endpoints/sequences.py b/src/app/api/api_v1/endpoints/sequences.py
index fd66d9cf..bdf9a752 100644
--- a/src/app/api/api_v1/endpoints/sequences.py
+++ b/src/app/api/api_v1/endpoints/sequences.py
@@ -7,7 +7,7 @@
 from datetime import date, timedelta
 from typing import Any, List, Union, cast
 
-from fastapi import APIRouter, Depends, HTTPException, Path, Query, Security, status
+from fastapi import APIRouter, Depends, HTTPException, Path, Query, Response, Security, status
 from sqlmodel import delete, func, select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
@@ -17,7 +17,7 @@
 from app.db import get_session
 from app.models import AlertSequence, AnnotationType, Camera, Detection, Sequence, UserRole
 from app.schemas.alerts import AlertCreate
-from app.schemas.detections import DetectionRead, DetectionSequence, DetectionWithUrl
+from app.schemas.detections import DetectionSequence, DetectionWithUrl
 from app.schemas.login import TokenPayload
 from app.schemas.sequences import SequenceLabel, SequenceRead
 from app.services.alerts import refresh_alert_state
@@ -29,6 +29,11 @@
 
 router = APIRouter()
 
+# Historical default, kept for sampling=1 so unsampled callers see no change.
+DEFAULT_DETECTION_LIMIT = 10
+# Ceiling on one response, mirrored in the limit Query constraint.
+MAX_DETECTION_LIMIT = 500
+
 
 async def verify_org_rights(
     organization_id: int, camera_id: int, cameras: CameraCRUD = Depends(get_camera_crud)
@@ -71,10 +76,38 @@ async def get_sequence(
     ),
 )
 async def fetch_sequence_detections(
+    response: Response,
     sequence_id: int = Path(..., gt=0),
-    limit: int = Query(10, description="Maximum number of detections to fetch", ge=1, le=100),
-    offset: int = Query(0, description="Number of detections to skip", ge=0),
+    limit: Union[int, None] = Query(
+        None,
+        description=(
+            "Maximum number of detections to fetch. Defaults to 10, except when `sampling` is set "
+            "and `limit` is omitted: it then defaults to whatever spans the whole sampled set "
+            "(capped at 500), so `?sampling=10` returns a spread instead of a truncated tail."
+        ),
+        ge=1,
+        le=500,
+    ),
+    offset: int = Query(0, description="Number of detections to skip, within the sampled set", ge=0),
     desc: bool = Query(True, description="Whether to order the detections by created_at in descending order"),
+    sampling: int = Query(
+        1,
+        description=(
+            "Keep one detection every N (1 = every detection). The kept frames are picked "
+            "chronologically from the start of the sequence, so the set does not depend on `desc` "
+            "and does not shift as the sequence grows; the first detection is always kept. "
+            "`limit` and `offset` then page that sampled set. Note that with `desc=true` a newly "
+            "recorded detection shifts page boundaries, since it lands at the front. "
+            "`sampling` thins the set and `limit` caps how much of it comes back, so an explicit "
+            "`limit` below `ceil(detections_count / sampling)` returns only part of the span (the "
+            "most recent part when `desc=true`). Omit `limit` to get the whole span, and read the "
+            "`X-Sampled-Total` and `X-Sampled-Truncated` response headers to tell whether what you "
+            "got covers the sequence. Since `limit` caps at 500, spanning a sequence in one call "
+            "needs `sampling >= detections_count / 500`."
+        ),
+        ge=1,
+        le=10_000,
+    ),
     with_crop: bool = Query(
         False,
         description="If true, presign and include crop_url for detections that have a crop. Defaults to false to skip the extra S3 head requests when crops are not needed.",
@@ -82,6 +115,7 @@ async def fetch_sequence_detections(
     cameras: CameraCRUD = Depends(get_camera_crud),
     detections: DetectionCRUD = Depends(get_detection_crud),
     sequences: SequenceCRUD = Depends(get_sequence_crud),
+    session: AsyncSession = Depends(get_session),
     token_payload: TokenPayload = Security(get_jwt, scopes=[UserRole.ADMIN, UserRole.AGENT, UserRole.USER]),
 ) -> List[DetectionWithUrl]:
     telemetry_client.capture(token_payload.sub, event="sequences-get", properties={"sequence_id": sequence_id})
@@ -90,18 +124,28 @@ async def fetch_sequence_detections(
     if not token_payload.is_admin and token_payload.organization_id != camera.organization_id:
         raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access forbidden.")
 
+    effective_limit = DEFAULT_DETECTION_LIMIT if limit is None else limit
+    if sampling > 1:
+        # Sampling thins the set but limit still truncates it, and truncation is invisible in the
+        # response: you get frames that look like a spread but are only its tail. So size the
+        # sampled set once (the same count the endpoint returns, continuity rows included), use it
+        # to fill in an omitted limit, and report it either way so a caller passing an explicit
+        # limit can still tell whether it covered the sequence.
+        counts = await get_detection_counts_by_sequence_ids(session, [sequence_id])
+        sampled_total = -(-counts.get(sequence_id, 0) // sampling)  # ceil division
+        if limit is None:
+            effective_limit = max(1, min(MAX_DETECTION_LIMIT, sampled_total))
+        response.headers["X-Sampled-Total"] = str(sampled_total)
+        response.headers["X-Sampled-Truncated"] = str(offset + effective_limit < sampled_total).lower()
+
     # Get the bucket of the camera's organization
     bucket = s3_service.get_bucket(s3_service.resolve_bucket_name(camera.organization_id))
-    fetched = await detections.fetch_all(
-        filters=("sequence_id", sequence_id),
-        order_by="created_at",
-        order_desc=desc,
-        limit=limit,
-        offset=offset,
+    fetched = await detections.fetch_by_sequence(
+        sequence_id, sampling=sampling, order_desc=desc, limit=effective_limit, offset=offset
     )
     return [
         DetectionWithUrl(
-            **DetectionRead(**elt.model_dump()).model_dump(),
+            **elt.model_dump(),
             url=bucket.get_public_url(elt.bucket_key, verify_exists=False),
             crop_url=(
                 bucket.get_public_url(elt.crop_bucket_key, verify_exists=False)
diff --git a/src/app/crud/crud_detection.py b/src/app/crud/crud_detection.py
index 3de161c7..1b7a35d4 100644
--- a/src/app/crud/crud_detection.py
+++ b/src/app/crud/crud_detection.py
@@ -3,9 +3,11 @@
 # This program is licensed under the Apache License 2.0.
 # See LICENSE or go to <https://www.apache.org/licenses/LICENSE-2.0> for full license details.
 
-from typing import Any, Union, cast
+from typing import Any, List, Union, cast
 
-from sqlalchemy import desc
+from sqlalchemy import desc, func
+from sqlalchemy import select as select_sa
+from sqlalchemy.orm import aliased
 from sqlmodel import select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
@@ -31,3 +33,60 @@ async def get_latest_with_bbox(self, sequence_id: int) -> Union[Detection, None]
         )
         results = await self.session.exec(statement)
         return results.first()
+
+    async def fetch_by_sequence(
+        self,
+        sequence_id: int,
+        sampling: int = 1,
+        order_desc: bool = True,
+        limit: int = 10,
+        offset: int = 0,
+    ) -> List[Detection]:
+        """Fetch the detections of a sequence, keeping one every ``sampling``.
+
+        The row number is always computed ascending on ``created_at``, so the sampled frame set
+        is the same whatever ``order_desc`` is: the latter only flips the output order. That
+        also keeps the set stable as the sequence grows, since a new detection lands last and
+        cannot renumber earlier rows, so the player keeps hitting the same frames (and the same
+        cached URLs) across polls. ``(rn - 1) % sampling == 0`` keeps the first detection, so a
+        non-empty sequence always yields at least one row. ``limit``/``offset`` page the
+        *sampled* set, not the raw rows.
+
+        Note that ``limit`` cannot push down: ``row_number()`` has to cover every row of the
+        sequence before the modulo and the limit apply, which is why detections is indexed on
+        ``(sequence_id, created_at)``.
+        """
+        if sampling <= 1:
+            # Unchanged pre-sampling behaviour, on purpose: same query, same ordering.
+            return await self.fetch_all(
+                filters=("sequence_id", sequence_id),
+                order_by="created_at",
+                order_desc=order_desc,
+                limit=limit,
+                offset=offset,
+            )
+
+        # id breaks created_at ties so the sampled set is deterministic run to run.
+        row_num = func.row_number().over(
+            order_by=(cast(Any, Detection.created_at).asc(), cast(Any, Detection.id).asc())
+        )
+        # sqlalchemy's select for the numbering subquery (two entities, and .subquery() on it);
+        # sqlmodel's select for the outer one, since a single-entity SelectOfScalar is what makes
+        # session.exec return Detection instances rather than Row tuples.
+        numbered: Any = select_sa(Detection, row_num.label("rn")).where(cast(Any, Detection.sequence_id) == sequence_id)
+        subq = numbered.subquery()
+        sampled = aliased(Detection, subq)
+        created_at_col = cast(Any, sampled.created_at)
+        id_col = cast(Any, sampled.id)
+        stmt: Any = (
+            select(sampled)
+            .where((subq.c.rn - 1) % sampling == 0)
+            .order_by(
+                created_at_col.desc() if order_desc else created_at_col.asc(),
+                id_col.desc() if order_desc else id_col.asc(),
+            )
+            .limit(limit)
+            .offset(offset)
+        )
+        result = await self.session.exec(stmt)
+        return list(result.all())
diff --git a/src/app/services/storage.py b/src/app/services/storage.py
index 45bc746e..47d2dd6e 100644
--- a/src/app/services/storage.py
+++ b/src/app/services/storage.py
@@ -5,8 +5,10 @@
 
 import hashlib
 import logging
+import time
+from collections import OrderedDict
 from mimetypes import guess_extension
-from typing import Any, BinaryIO, Dict, Union
+from typing import Any, BinaryIO, Dict, Tuple, Union
 
 import boto3
 import magic
@@ -21,6 +23,21 @@
 
 logger = logging.getLogger("uvicorn.warning")
 
+# Presigned URLs run 600-900 bytes, so ~1 KB per entry => ~8 MB hard ceiling per bucket.
+_URL_CACHE_MAXSIZE = 8192
+
+
+def _url_cache_window(url_expiration: int) -> int:
+    """How long a single presigned URL string keeps being handed out, in seconds.
+
+    Derived from the expiration rather than hardcoded, so lowering S3_URL_EXPIRATION can never
+    start serving already-expired URLs: the window never exceeds the lifetime, whatever the
+    input (the floor is 1, not 60, so a sub-minute expiration degrades to near-no-caching instead
+    of outliving the URL). At the 24h default the cap binds instead and the window is 1h, so a
+    handed-out URL always has >= 23h left.
+    """
+    return min(3600, max(1, url_expiration // 4))
+
 
 class S3Bucket:
     """S3 bucket manager
@@ -41,6 +58,9 @@ def __init__(self, s3_client, bucket_name: str, proxy_url: Union[str, None] = No
             raise ValueError(f"unable to access bucket {bucket_name}")
         self.name = bucket_name
         self.proxy_url = proxy_url
+        # (bucket_key, url_expiration, window slot) -> presigned URL. Scoped to the instance
+        # because proxy_url and the signing credentials are fixed per bucket.
+        self._url_cache: OrderedDict[Tuple[str, int, int], str] = OrderedDict()
 
     def get_file_metadata(self, bucket_key: str) -> Dict[str, Any]:
         # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html#S3.Client.head_object
@@ -59,7 +79,16 @@ def check_file_existence(self, bucket_key: str) -> bool:
     def upload_file(self, bucket_key: str, file_binary: BinaryIO) -> bool:
         """Upload a file to bucket and return whether the upload succeeded"""
         # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html#S3.Bucket.upload_fileobj
-        self._s3.upload_fileobj(file_binary, self.name, bucket_key)
+        # Cache-Control is what turns the stable presigned URLs from get_public_url into actual
+        # browser cache hits: without it browsers only cache heuristically, so the player
+        # re-downloads frames it already has. Objects are immutable once written (the key
+        # embeds a content hash), so max-age can match the URL lifetime.
+        self._s3.upload_fileobj(
+            file_binary,
+            self.name,
+            bucket_key,
+            ExtraArgs={"CacheControl": f"private, max-age={settings.S3_URL_EXPIRATION}"},
+        )
         return True
 
     def delete_file(self, bucket_key: str) -> None:
@@ -82,12 +111,17 @@ def get_public_url(
                 exist (e.g. sequence detections): the client then gets a 403/404 from S3
                 when loading the URL instead of an upfront error.
         """
+        # Checked before the cache lookup on purpose: a cache hit must not skip the existence
+        # check callers opted into.
         if verify_exists and not self.check_file_existence(bucket_key):
             raise HTTPException(
                 status_code=status.HTTP_404_NOT_FOUND, detail="File cannot be found on the bucket storage"
             )
 
-        # Generate a public URL for it using boto3 presign URL generation\
+        return self._stable_presign(bucket_key, url_expiration)
+
+    def _presign(self, bucket_key: str, url_expiration: int) -> str:
+        # Generate a public URL for it using boto3 presign URL generation
         presigned_url = self._s3.generate_presigned_url(
             "get_object", Params={"Bucket": self.name, "Key": bucket_key}, ExpiresIn=url_expiration
         )
@@ -95,6 +129,46 @@ def get_public_url(
             return presigned_url.replace(self._s3.meta.endpoint_url, self.proxy_url)
         return presigned_url
 
+    def _stable_presign(self, bucket_key: str, url_expiration: int) -> str:
+        """Return the same URL string for a whole window, so the browser can cache frames.
+
+        boto3 stamps the current clock into every signature (``X-Amz-Date`` under SigV4,
+        ``Expires`` under SigV2), so re-presigning the same key yields a different string and
+        busts the client cache on each poll. The browser keys its cache on the full URL.
+
+        The window slot is part of the cache key, so a lookup against the current slot can never
+        return a previous window's entry: no explicit rollover clear is needed, and stale entries
+        simply age out through the size-bound eviction below.
+
+        Stability is per-process: the cache lives on this instance, so it only dedupes URLs
+        within one worker. With W workers a polling client sees up to W distinct URLs per key
+        per window, one per worker that happened to answer, so a frame is fetched up to W times
+        instead of once. That is still far better than re-signing on every poll (which bust the
+        cache unconditionally), just divided by W. Production runs several workers, so this
+        applies there; the worker count is not visible from this repo, which only carries dev
+        compose files. See #671.
+
+        Making it cross-process is not achievable without patching boto3's clock: SigV4 derives
+        ``X-Amz-Date`` from the wall clock at signing time rather than taking it as a
+        ``generate_presigned_url`` parameter, so a shared cache would be the way in.
+        """
+        window = _url_cache_window(url_expiration)
+        # monotonic: only the window length matters, and it is immune to NTP steps.
+        slot = int(time.monotonic()) // window
+        cache_key = (bucket_key, url_expiration, slot)
+        url = self._url_cache.get(cache_key)
+        if url is not None:
+            self._url_cache.move_to_end(cache_key)
+            return url
+        url = self._presign(bucket_key, url_expiration)
+        if len(self._url_cache) >= _URL_CACHE_MAXSIZE:
+            # Evict the coldest entry, never clear: one dict is shared by every viewer of an
+            # organization, so clearing here would re-presign (and so change) every URL in
+            # flight exactly when the cache is under load and stability matters most.
+            self._url_cache.popitem(last=False)
+        self._url_cache[cache_key] = url
+        return url
+
     async def delete_items(self) -> None:
         """Delete all items in the bucket"""
         paginator = self._s3.get_paginator("list_objects_v2")
@@ -131,6 +205,10 @@ def __init__(
             raise ValueError("unable to access S3")
         logger.info(f"S3 connected on {endpoint_url}")
         self.proxy_url = proxy_url
+        # bucket_name -> S3Bucket. S3Bucket.__init__ does a blocking head_bucket round-trip on
+        # the event loop; the bucket set is one per organization and effectively static, so
+        # build each one once. Caching the instance is also what lets its URL cache ever hit.
+        self._buckets: Dict[str, S3Bucket] = {}
 
     def create_bucket(self, bucket_name: str) -> bool:
         """Create a new bucket in S3 storage"""
@@ -149,19 +227,31 @@ def create_bucket(self, bucket_name: str) -> bool:
             return False
 
     def get_bucket(self, bucket_name: str) -> S3Bucket:
-        """Get an existing bucket in S3 storage"""
-        return S3Bucket(self._s3, bucket_name, self.proxy_url)
+        """Get an existing bucket in S3 storage (cached instance)
+
+        Only successful lookups are cached, so a missing bucket still raises ValueError. Once
+        cached, a bucket deleted out-of-band keeps answering and the failure surfaces at the S3
+        call rather than here.
+        """
+        bucket = self._buckets.get(bucket_name)
+        if bucket is None:
+            bucket = S3Bucket(self._s3, bucket_name, self.proxy_url)
+            self._buckets[bucket_name] = bucket
+        return bucket
 
     async def delete_bucket(self, bucket_name: str) -> bool:
         """Delete an existing bucket in S3 storage"""
-        bucket = S3Bucket(self._s3, bucket_name, self.proxy_url)
+        bucket = self.get_bucket(bucket_name)
         try:
             await bucket.delete_items()
             self._s3.delete_bucket(Bucket=bucket_name)
-            return True
         except ClientError as e:
             logger.warning(e)
             return False
+        # Evict only once the bucket is really gone, so the cache stays coherent with the
+        # organization-deletion path.
+        self._buckets.pop(bucket_name, None)
+        return True
 
     @staticmethod
     def resolve_bucket_name(organization_id: int) -> str:
diff --git a/src/tests/conftest.py b/src/tests/conftest.py
index a6203e73..b8b40905 100644
--- a/src/tests/conftest.py
+++ b/src/tests/conftest.py
@@ -22,6 +22,7 @@
 from app.db import engine, session_factory
 from app.main import app
 from app.models import Camera, Detection, OcclusionMask, Organization, Pose, Sequence, User, Webhook
+from app.services import storage
 from app.services.storage import s3_service
 from app.services.validation import process_next_due_validation
 
@@ -338,6 +339,21 @@ def mock_img():
     return requests.get("https://avatars.githubusercontent.com/u/61667887?s=200&v=4", timeout=5).content
 
 
+@pytest.fixture
+def pinned_url_window(monkeypatch):
+    """Freeze the presigned-URL cache window so a test cannot straddle a rollover.
+
+    Any assertion that two requests return the same url is otherwise wall-clock dependent: it
+    fails whenever the requests land either side of a window boundary. A window far longer than
+    any process uptime keeps the slot constant instead.
+    """
+    monkeypatch.setattr(storage, "_url_cache_window", lambda _url_expiration: 10**9)
+    for bucket in storage.s3_service._buckets.values():
+        # Both fixture consumers pin the window to the same huge value, so without clearing,
+        # whichever test runs second would start warm off the other's cache entries.
+        bucket._url_cache.clear()
+
+
 @pytest_asyncio.fixture(loop_scope="session")
 async def organization_session(async_session: AsyncSession):
     for entry in ORGANIZATION_TABLE:
diff --git a/src/tests/endpoints/test_sequences.py b/src/tests/endpoints/test_sequences.py
index 1cb5d5f7..4e258823 100644
--- a/src/tests/endpoints/test_sequences.py
+++ b/src/tests/endpoints/test_sequences.py
@@ -899,6 +899,361 @@ async def test_fetch_sequence_detections_offset_validation(
     assert response.status_code == 422
 
 
+async def _seed_sampling_sequence(session: AsyncSession, count: int = 10) -> tuple[int, List[int]]:
+    """Create a sequence with `count` detections one minute apart, oldest first.
+
+    Sequence 1 in the fixtures only has 3 detections, which is too few to tell an interval
+    apart from a limit. Returns the sequence id and the detection ids in chronological order.
+    """
+    now = utcnow()
+    sequence = Sequence(
+        camera_id=pytest.camera_table[0]["id"],
+        pose_id=pytest.pose_table[0]["id"],
+        camera_azimuth=180.0,
+        sequence_azimuth=175.0,
+        cone_angle=5.0,
+        is_wildfire=None,
+        started_at=now - timedelta(minutes=count),
+        last_seen_at=now,
+    )
+    session.add(sequence)
+    await session.commit()
+    await session.refresh(sequence)
+
+    detections = [
+        Detection(
+            camera_id=sequence.camera_id,
+            pose_id=pytest.pose_table[0]["id"],
+            sequence_id=sequence.id,
+            bucket_key=f"sampling-{sequence.id}-{idx}.jpg",
+            bbox="[(.1,.1,.7,.8,.9)]",
+            others_bboxes=None,
+            created_at=now - timedelta(minutes=count - idx),
+        )
+        for idx in range(count)
+    ]
+    for detection in detections:
+        session.add(detection)
+    await session.commit()
+    for detection in detections:
+        await session.refresh(detection)
+
+    return sequence.id, [detection.id for detection in detections]
+
+
+@pytest.mark.asyncio
+async def test_fetch_sequence_detections_sampling_set_is_independent_of_desc(
+    async_client: AsyncClient,
+    detection_session: AsyncSession,
+):
+    """The sampled frame set must not depend on `desc`: same ids, reversed order.
+
+    This is the whole point of numbering rows ascending regardless of the output order. If it
+    regresses, the player shows different frames depending on scroll direction.
+    """
+    sequence_id, _ = await _seed_sampling_sequence(detection_session)
+    auth = pytest.get_token(
+        pytest.user_table[0]["id"],
+        pytest.user_table[0]["role"].split(),
+        pytest.user_table[0]["organization_id"],
+    )
+
+    asc = await async_client.get(f"/sequences/{sequence_id}/detections?sampling=2&desc=false&limit=10", headers=auth)
+    desc = await async_client.get(f"/sequences/{sequence_id}/detections?sampling=2&desc=true&limit=10", headers=auth)
+    assert asc.status_code == 200, asc.text
+    assert desc.status_code == 200, desc.text
+
+    asc_ids = [det["id"] for det in asc.json()]
+    desc_ids = [det["id"] for det in desc.json()]
+    assert asc_ids == list(reversed(desc_ids))
+
+
+@pytest.mark.asyncio
+async def test_fetch_sequence_detections_sampling_one_matches_default(
+    async_client: AsyncClient,
+    detection_session: AsyncSession,
+    pinned_url_window: None,
+):
+    """sampling=1 must be indistinguishable from not passing it at all, urls included."""
+    sequence_id, _ = await _seed_sampling_sequence(detection_session)
+    auth = pytest.get_token(
+        pytest.user_table[0]["id"],
+        pytest.user_table[0]["role"].split(),
+        pytest.user_table[0]["organization_id"],
+    )
+
+    default = await async_client.get(f"/sequences/{sequence_id}/detections?limit=10&desc=false", headers=auth)
+    sampled = await async_client.get(
+        f"/sequences/{sequence_id}/detections?limit=10&desc=false&sampling=1", headers=auth
+    )
+    assert default.status_code == 200, default.text
+    assert sampled.status_code == 200, sampled.text
+    # Full equality, urls included. The fixture pins the presign window so a rollover between
+    # the two requests can't make the urls differ for reasons unrelated to sampling; that the
+    # cache is actually used is covered by
+    # test_fetch_sequence_detections_reuses_presigned_urls_across_requests.
+    assert default.json() == sampled.json()
+
+
+@pytest.mark.asyncio
+async def test_fetch_sequence_detections_sampling_interval(
+    async_client: AsyncClient,
+    detection_session: AsyncSession,
+):
+    """sampling=2 over 10 detections keeps chronological positions 1, 3, 5, 7, 9."""
+    sequence_id, chronological_ids = await _seed_sampling_sequence(detection_session)
+    auth = pytest.get_token(
+        pytest.user_table[0]["id"],
+        pytest.user_table[0]["role"].split(),
+        pytest.user_table[0]["organization_id"],
+    )
+
+    response = await async_client.get(
+        f"/sequences/{sequence_id}/detections?sampling=2&desc=false&limit=10", headers=auth
+    )
+    assert response.status_code == 200, response.text
+    assert [det["id"] for det in response.json()] == chronological_ids[::2]
+
+
+@pytest.mark.asyncio
+async def test_fetch_sequence_detections_omitted_limit_spans_the_sampled_set(
+    async_client: AsyncClient,
+    detection_session: AsyncSession,
+):
+    """Without an explicit limit, sampling must return the whole span, not its tail.
+
+    This is the guardrail: with the historical limit=10 default, `?sampling=2` on a 30-detection
+    sequence would silently return the 10 most recent sampled frames while looking like a spread.
+    """
+    sequence_id, chronological_ids = await _seed_sampling_sequence(detection_session, count=30)
+    auth = pytest.get_token(
+        pytest.user_table[0]["id"],
+        pytest.user_table[0]["role"].split(),
+        pytest.user_table[0]["organization_id"],
+    )
+
+    response = await async_client.get(f"/sequences/{sequence_id}/detections?sampling=2&desc=false", headers=auth)
+    assert response.status_code == 200, response.text
+    # ceil(30 / 2) = 15 frames, spanning the sequence rather than stopping at 10.
+    assert [det["id"] for det in response.json()] == chronological_ids[::2]
+    assert len(response.json()) == 15
+    assert response.headers["x-sampled-total"] == "15"
+    assert response.headers["x-sampled-truncated"] == "false"
+
+
+@pytest.mark.asyncio
+async def test_fetch_sequence_detections_explicit_limit_still_wins_and_reports_truncation(
+    async_client: AsyncClient,
+    detection_session: AsyncSession,
+):
+    """An explicit limit is honoured, and the headers say the span was cut short."""
+    sequence_id, chronological_ids = await _seed_sampling_sequence(detection_session, count=30)
+    auth = pytest.get_token(
+        pytest.user_table[0]["id"],
+        pytest.user_table[0]["role"].split(),
+        pytest.user_table[0]["organization_id"],
+    )
+
+    response = await async_client.get(
+        f"/sequences/{sequence_id}/detections?sampling=2&desc=false&limit=4", headers=auth
+    )
+    assert response.status_code == 200, response.text
+    assert [det["id"] for det in response.json()] == chronological_ids[::2][:4]
+    assert response.headers["x-sampled-total"] == "15"
+    assert response.headers["x-sampled-truncated"] == "true"
+
+    # Paging to the end of the sampled set is not truncation.
+    tail = await async_client.get(
+        f"/sequences/{sequence_id}/detections?sampling=2&desc=false&limit=4&offset=11", headers=auth
+    )
+    assert tail.status_code == 200, tail.text
+    assert tail.headers["x-sampled-truncated"] == "false"
+
+
+@pytest.mark.asyncio
+async def test_fetch_sequence_detections_unsampled_default_limit_unchanged(
+    async_client: AsyncClient,
+    detection_session: AsyncSession,
+):
+    """sampling=1 keeps the historical limit=10 default and skips the extra count entirely."""
+    sequence_id, chronological_ids = await _seed_sampling_sequence(detection_session, count=30)
+    auth = pytest.get_token(
+        pytest.user_table[0]["id"],
+        pytest.user_table[0]["role"].split(),
+        pytest.user_table[0]["organization_id"],
+    )
+
+    response = await async_client.get(f"/sequences/{sequence_id}/detections?desc=false", headers=auth)
+    assert response.status_code == 200, response.text
+    assert [det["id"] for det in response.json()] == chronological_ids[:10]
+    # The headers describe a sampled set, so they are absent when nothing was sampled.
+    assert "x-sampled-total" not in response.headers
+
+
+@pytest.mark.asyncio
+async def test_fetch_sequence_detections_omitted_limit_handles_degenerate_sampling(
+    async_client: AsyncClient,
+    detection_session: AsyncSession,
+):
+    """A sampling interval past the detection count still yields one row, not zero."""
+    sequence_id, chronological_ids = await _seed_sampling_sequence(detection_session, count=10)
+    auth = pytest.get_token(
+        pytest.user_table[0]["id"],
+        pytest.user_table[0]["role"].split(),
+        pytest.user_table[0]["organization_id"],
+    )
+
+    response = await async_client.get(f"/sequences/{sequence_id}/detections?sampling=500", headers=auth)
+    assert response.status_code == 200, response.text
+    assert [det["id"] for det in response.json()] == [chronological_ids[0]]
+    assert response.headers["x-sampled-total"] == "1"
+    assert response.headers["x-sampled-truncated"] == "false"
+
+
+@pytest.mark.asyncio
+async def test_fetch_sequence_detections_sampling_larger_than_count(
+    async_client: AsyncClient,
+    detection_session: AsyncSession,
+):
+    """A sampling interval above the detection count still returns the first detection."""
+    sequence_id, chronological_ids = await _seed_sampling_sequence(detection_session)
+    auth = pytest.get_token(
+        pytest.user_table[0]["id"],
+        pytest.user_table[0]["role"].split(),
+        pytest.user_table[0]["organization_id"],
+    )
+
+    response = await async_client.get(
+        f"/sequences/{sequence_id}/detections?sampling=100&desc=false&limit=10", headers=auth
+    )
+    assert response.status_code == 200, response.text
+    assert [det["id"] for det in response.json()] == [chronological_ids[0]]
+
+
+@pytest.mark.asyncio
+async def test_fetch_sequence_detections_sampling_offset_applies_to_sampled_set(
+    async_client: AsyncClient,
+    detection_session: AsyncSession,
+):
+    """offset skips sampled rows, not raw rows: positions 5 and 7, not 3 and 4."""
+    sequence_id, chronological_ids = await _seed_sampling_sequence(detection_session)
+    auth = pytest.get_token(
+        pytest.user_table[0]["id"],
+        pytest.user_table[0]["role"].split(),
+        pytest.user_table[0]["organization_id"],
+    )
+
+    response = await async_client.get(
+        f"/sequences/{sequence_id}/detections?sampling=2&desc=false&limit=2&offset=2", headers=auth
+    )
+    assert response.status_code == 200, response.text
+    assert [det["id"] for det in response.json()] == [chronological_ids[4], chronological_ids[6]]
+
+
+@pytest.mark.asyncio
+async def test_fetch_sequence_detections_sampling_offset_desc_paging(
+    async_client: AsyncClient,
+    detection_session: AsyncSession,
+):
+    """`desc=true` only reverses the OUTPUT order of the sampled set, offset then skips
+    from the front of that reversed list.
+
+    The sampled set is chosen by ascending created_at: positions 1,3,5,7,9, i.e.
+    chronological_ids[::2] = [c0, c2, c4, c6, c8]. Reversing for desc=true gives
+    [c8, c6, c4, c2, c0]; offset=1, limit=2 then skips c8 and returns [c6, c4].
+    """
+    sequence_id, chronological_ids = await _seed_sampling_sequence(detection_session)
+    auth = pytest.get_token(
+        pytest.user_table[0]["id"],
+        pytest.user_table[0]["role"].split(),
+        pytest.user_table[0]["organization_id"],
+    )
+
+    response = await async_client.get(
+        f"/sequences/{sequence_id}/detections?sampling=2&desc=true&limit=2&offset=1", headers=auth
+    )
+    assert response.status_code == 200, response.text
+    assert [det["id"] for det in response.json()] == [chronological_ids[6], chronological_ids[4]]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("query", "status_code"),
+    [
+        ("sampling=0", 422),
+        ("sampling=-1", 422),
+        ("sampling=1", 200),
+        ("sampling=10000", 200),
+        ("sampling=10001", 422),
+        # asyncpg can't encode this into int8, so without an upper bound this reached the
+        # database and came back as a 500 instead of a validation error.
+        ("sampling=9223372036854775808", 422),
+        ("limit=500", 200),
+        ("limit=501", 422),
+    ],
+)
+async def test_fetch_sequence_detections_sampling_validation(
+    async_client: AsyncClient,
+    detection_session: AsyncSession,
+    query: str,
+    status_code: int,
+):
+    auth = pytest.get_token(
+        pytest.user_table[0]["id"],
+        pytest.user_table[0]["role"].split(),
+        pytest.user_table[0]["organization_id"],
+    )
+    response = await async_client.get(f"/sequences/1/detections?{query}", headers=auth)
+    assert response.status_code == status_code, response.text
+
+
+@pytest.mark.asyncio
+async def test_fetch_sequence_detections_reuses_presigned_urls_across_requests(
+    async_client: AsyncClient,
+    detection_session: AsyncSession,
+    pinned_url_window: None,
+):
+    """A repeat request must serve cached urls instead of re-signing every key.
+
+    boto3 stamps the current clock into every signature, so without the presign cache the
+    browser's cache key changes on every poll and the player re-downloads every frame. Asserting
+    the urls merely match is not enough: two signatures taken in the same wall-clock second are
+    identical anyway, so the presign calls are counted instead.
+    """
+    auth = pytest.get_token(
+        pytest.user_table[0]["id"],
+        pytest.user_table[0]["role"].split(),
+        pytest.user_table[0]["organization_id"],
+    )
+    bucket = s3_service.get_bucket(s3_service.resolve_bucket_name(pytest.camera_table[0]["organization_id"]))
+    presign_calls: List[str] = []
+    original_presign = bucket._presign
+
+    def counting_presign(bucket_key: str, url_expiration: int) -> str:
+        presign_calls.append(bucket_key)
+        return original_presign(bucket_key, url_expiration)
+
+    bucket._presign = counting_presign  # type: ignore[method-assign]
+    try:
+        first = await async_client.get("/sequences/1/detections?limit=10&desc=false", headers=auth)
+        signed_once = len(presign_calls)
+        second = await async_client.get("/sequences/1/detections?limit=10&desc=false", headers=auth)
+    finally:
+        del bucket._presign
+
+    assert first.status_code == 200, first.text
+    assert second.status_code == 200, second.text
+
+    first_urls = {det["id"]: det["url"] for det in first.json()}
+    assert first_urls
+    assert {det["id"]: det["url"] for det in second.json()} == first_urls
+    # One presign per distinct key: the fixture detections of sequence 1 all share a bucket_key,
+    # so the cache already collapses them within a single request.
+    assert signed_once == len(set(first_urls.values()))
+    # The second request signed nothing at all.
+    assert len(presign_calls) == signed_once
+
+
 @pytest.mark.asyncio
 async def test_unit_label_sequence_forbidden_for_wrong_org():
     """Verify that an AGENT from a different organization cannot label the sequence."""
diff --git a/src/tests/services/test_storage.py b/src/tests/services/test_storage.py
index 04c0f7fc..bd7b8fe4 100644
--- a/src/tests/services/test_storage.py
+++ b/src/tests/services/test_storage.py
@@ -5,7 +5,8 @@
 from fastapi import HTTPException
 
 from app.core.config import settings
-from app.services.storage import S3Bucket, S3Service
+from app.services import storage
+from app.services.storage import S3Bucket, S3Service, _url_cache_window
 
 
 @pytest.mark.parametrize(
@@ -86,6 +87,9 @@ async def test_s3_bucket(bucket_name, proxy_url, expected_error, mock_img):
         bucket.upload_file(bucket_key, io.BytesIO(mock_img))
         assert bucket.check_file_existence(bucket_key)
         assert isinstance(bucket.get_file_metadata(bucket_key), dict)
+        # Cache-Control is what makes the stable presigned urls from get_public_url actually
+        # cacheable browser-side; without it the upload path could silently stop setting it.
+        assert bucket.get_file_metadata(bucket_key)["CacheControl"] == f"private, max-age={settings.S3_URL_EXPIRATION}"
         assert bucket.get_public_url(bucket_key).startswith("http://")
         # Delete file
         bucket.delete_file(bucket_key)
@@ -100,3 +104,115 @@ async def test_s3_bucket(bucket_name, proxy_url, expected_error, mock_img):
     else:
         with pytest.raises(expected_error):
             S3Bucket(s3, bucket_name, proxy_url)
+
+
+@pytest.mark.parametrize("url_expiration", [1, 4, 20, 60, 300, 3600, 24 * 3600])
+def test_url_cache_window_leaves_most_of_the_lifetime(url_expiration):
+    """A cached url must always be handed out with the bulk of its lifetime left, and the window
+    must never exceed the expiration itself: a sub-minute expiration must not serve urls that are
+    already expired."""
+    window = _url_cache_window(url_expiration)
+    assert 1 <= window <= 3600
+    assert window <= max(1, url_expiration // 4)
+    assert window <= url_expiration
+
+
+def test_s3_bucket_presigned_urls_are_stable_within_a_window(monkeypatch):
+    """The same key presigns once per window, and is re-signed once the slot advances.
+
+    boto3 stamps the current clock into every signature, so without this cache the browser's
+    cache key changes on every request and clients re-download unchanged objects. Both halves
+    count presign calls rather than comparing url strings: two signatures taken in the same
+    wall-clock second are identical, so a string comparison would assert on the clock instead of
+    on the cache. The clock is faked, rather than waited out or mutated on the instance (the
+    window slot now lives in the cache key, not on a `bucket._url_window` attribute), so the
+    slot advance is deterministic instead of racing a real monotonic boundary.
+    """
+    session = boto3.Session(settings.S3_ACCESS_KEY, settings.S3_SECRET_KEY, region_name=settings.S3_REGION)
+    s3 = session.client("s3", endpoint_url=settings.S3_ENDPOINT_URL)
+    bucket_name = "dummy-bucket-url-cache"
+    s3.create_bucket(Bucket=bucket_name, CreateBucketConfiguration={"LocationConstraint": settings.S3_REGION})
+    try:
+        bucket = S3Bucket(s3, bucket_name, settings.S3_PROXY_URL)
+        presign_calls = []
+        original_presign = bucket._presign
+
+        def counting_presign(bucket_key, url_expiration):
+            presign_calls.append(bucket_key)
+            return original_presign(bucket_key, url_expiration)
+
+        bucket._presign = counting_presign
+
+        fake_clock = [0.0]
+        monkeypatch.setattr(storage.time, "monotonic", lambda: fake_clock[0])
+
+        first = bucket.get_public_url("stable.png", verify_exists=False)
+        assert bucket.get_public_url("stable.png", verify_exists=False) == first
+        assert len(presign_calls) == 1
+
+        # Advance the clock past the window instead of waiting one out: the slot changes, so the
+        # key changes, so the lookup misses and the url is re-signed.
+        fake_clock[0] += _url_cache_window(settings.S3_URL_EXPIRATION)
+        bucket.get_public_url("stable.png", verify_exists=False)
+        assert len(presign_calls) == 2
+        # Nothing clears on rollover anymore: the previous window's entry is still there,
+        # aging out through the size-bound eviction rather than being dropped outright.
+        assert set(bucket._url_cache) == {
+            ("stable.png", settings.S3_URL_EXPIRATION, 0),
+            ("stable.png", settings.S3_URL_EXPIRATION, 1),
+        }
+    finally:
+        s3.delete_bucket(Bucket=bucket_name)
+
+
+def test_s3_bucket_url_cache_evicts_coldest_entry_only(monkeypatch):
+    """The size bound evicts only the least-recently-used entry; it must never clear the whole
+    cache, since one dict is shared by every viewer of an organization and clearing it would
+    re-sign (and so change) every url in flight exactly when the cache is under load.
+    """
+    session = boto3.Session(settings.S3_ACCESS_KEY, settings.S3_SECRET_KEY, region_name=settings.S3_REGION)
+    s3 = session.client("s3", endpoint_url=settings.S3_ENDPOINT_URL)
+    bucket_name = "dummy-bucket-url-cache-eviction"
+    s3.create_bucket(Bucket=bucket_name, CreateBucketConfiguration={"LocationConstraint": settings.S3_REGION})
+    try:
+        bucket = S3Bucket(s3, bucket_name, settings.S3_PROXY_URL)
+        # Pin the window so the three calls below can't straddle a real rollover and pick up
+        # differing slot components in their cache keys.
+        monkeypatch.setattr(storage, "_url_cache_window", lambda _url_expiration: 10**9)
+        monkeypatch.setattr(storage, "_URL_CACHE_MAXSIZE", 2)
+
+        bucket.get_public_url("k1.png", verify_exists=False)
+        bucket.get_public_url("k2.png", verify_exists=False)
+        # Touch k1 again: it becomes most-recently-used, leaving k2 as the coldest entry.
+        bucket.get_public_url("k1.png", verify_exists=False)
+        # Inserting a third key over the bound of 2 must evict k2 only, never clear the dict.
+        bucket.get_public_url("k3.png", verify_exists=False)
+
+        cached_bucket_keys = {key[0] for key in bucket._url_cache}
+        assert cached_bucket_keys == {"k1.png", "k3.png"}
+        assert len(bucket._url_cache) == 2
+    finally:
+        s3.delete_bucket(Bucket=bucket_name)
+
+
+@pytest.mark.asyncio
+async def test_s3_service_caches_bucket_instances():
+    """get_bucket must reuse instances (its __init__ does a blocking head_bucket) and evict
+    the entry once the bucket is deleted."""
+    service = S3Service(
+        settings.S3_REGION,
+        settings.S3_ENDPOINT_URL,
+        settings.S3_ACCESS_KEY,
+        settings.S3_SECRET_KEY,
+        settings.S3_PROXY_URL,
+    )
+    bucket_name = "dummy-bucket-instance-cache"
+    service.create_bucket(bucket_name)
+    assert service.get_bucket(bucket_name) is service.get_bucket(bucket_name)
+
+    assert await service.delete_bucket(bucket_name)
+    assert bucket_name not in service._buckets
+
+    # A missing bucket is never cached, so it keeps raising.
+    with pytest.raises(ValueError, match="unable to access bucket"):
+        service.get_bucket("dummy-bucket-does-not-exist")