feat(remote): forward HLS segment filters to remote clients

Add POST /api/session/{id}/segment_filter: the server runs the service's OnSegmentFilter and returns only the unwanted segment URIs, so ads and bumpers are dropped over --remote-only without exposing service code.

docs/dev/rest-api/endpoints.md

@@ -826,6 +826,7 @@ GET  /api/session/{id}/prompt   → poll until authenticated (submit answers via
 GET  /api/session/{id}/titles   → list titles
 POST /api/session/{id}/tracks   → tracks + chapters (with URLs) for one title
 POST /api/session/{id}/segments → resolve segment URLs / DRM / headers per track
+POST /api/session/{id}/segment_filter → HLS segments the service drops (ads, bumpers)
 POST /api/session/{id}/license  → proxy or server-CDM DRM licensing
 DELETE /api/session/{id}        → tear down, harvest updated cache
 ```
@@ -964,6 +965,23 @@ Get the download URL, DRM info, headers, cookies, and raw track data for the sel
 | `400` | `INVALID_INPUT` | Missing `track_ids`. |
 | `404` | `TRACK_NOT_FOUND` | An unknown track ID was requested. |
 
+### `POST /api/session/{session_id}/segment_filter`
+
+Get the HLS segments a service drops (ads, bumpers, dub cards). The server runs the service's own segment filter over the track's media playlist and returns only the URIs, so the client drops the same segments the service does. Body: `{ "track_id": "v-1" }` (required). `unwanted` is `null` when the track has no filter. The client keeps every segment when the server does not have this route.
+
+=== "Response `200`"
+
+    ```json
+    { "unwanted": ["https://cdn.example/hls/ad1.ts"] }
+    ```
+
+| Status | Error code | Meaning |
+| --- | --- | --- |
+| `200` | - | Unwanted segment URIs, or `null`. |
+| `400` | `INVALID_INPUT` | Missing `track_id`. |
+| `404` | `TRACK_NOT_FOUND` | An unknown track ID was requested. |
+| `500` | `SERVICE_ERROR` | The playlist request or the filter failed. |
+
 ### `POST /api/session/{session_id}/license`
 
 Get the content keys for the DRM. The `mode` field selects one of two modes.

docs/dev/rest-api/remote-sessions.md

@@ -278,6 +278,7 @@ The server mounts all these routes, even in `--remote-only` mode. Paths use the
 | `GET` | `/api/session/{id}/titles` | Fetch and cache the title list |
 | `POST` | `/api/session/{id}/tracks` | Resolve tracks, manifests, chapters, headers/cookies for a title |
 | `POST` | `/api/session/{id}/segments` | Resolve per-segment/track download descriptors |
+| `POST` | `/api/session/{id}/segment_filter` | Unwanted HLS segment URIs for one track (ads, bumpers) |
 | `POST` | `/api/session/{id}/license` | DRM licensing (proxy or server CDM) |
 | `GET` | `/api/session/{id}/prompt` | Poll interactive auth status / pending prompt |
 | `POST` | `/api/session/{id}/prompt` | Submit an answer to a pending prompt |

tests/remote/unit/test_segment_filter.py

@@ -0,0 +1,92 @@
+"""The server runs a track's OnSegmentFilter and the client skips the segments it names."""
+
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+
+import pytest
+import responses
+
+from unshackle.core.api import handlers
+from unshackle.core.remote_service import RemoteClient, RemoteService
+
+pytestmark = pytest.mark.unit
+
+PLAYLIST = """#EXTM3U
+#EXT-X-TARGETDURATION:4
+#EXTINF:4,
+main0.ts
+#EXTINF:4,
+ad1.ts
+#EXTINF:4,
+main2.ts
+#EXT-X-ENDLIST
+"""
+
+BASE = "https://cdn.test/hls/media.m3u8"
+
+
+def make_session(monkeypatch, track):
+    session = SimpleNamespace(
+        tracks={"t1": track},
+        service_instance=SimpleNamespace(session=SimpleNamespace(get=lambda url: SimpleNamespace(text=PLAYLIST))),
+    )
+
+    async def get_validated_session(session_id, request):
+        return session
+
+    monkeypatch.setattr(handlers, "get_validated_session", get_validated_session)
+    return session
+
+
+async def test_handler_returns_unwanted_absolute_uris(monkeypatch):
+    track = SimpleNamespace(url=BASE, OnSegmentFilter=lambda segment: "ad" in segment.uri)
+    make_session(monkeypatch, track)
+
+    response = await handlers.session_segment_filter_handler({"track_id": "t1"}, "s1")
+
+    assert json.loads(response.body) == {"unwanted": ["https://cdn.test/hls/ad1.ts"]}
+
+
+async def test_handler_returns_null_without_a_filter(monkeypatch):
+    make_session(monkeypatch, SimpleNamespace(url=BASE, OnSegmentFilter=None))
+
+    response = await handlers.session_segment_filter_handler({"track_id": "t1"}, "s1")
+
+    assert json.loads(response.body) == {"unwanted": None}
+
+
+def make_service(unwanted, status=200):
+    service = RemoteService.__new__(RemoteService)
+    service.client = RemoteClient(server_url="http://srv:8786", api_key="k")
+    service._session_id = "s1"
+    service._segment_filters = {}
+    responses.add(
+        responses.POST,
+        "http://srv:8786/api/session/s1/segment_filter",
+        json={"unwanted": unwanted} if status == 200 else {"message": "no"},
+        status=status,
+    )
+    return service
+
+
+@responses.activate
+def test_client_filter_drops_only_the_unwanted_segments():
+    service = make_service(["https://cdn.test/hls/ad1.ts"])
+    segment_filter = service.remote_segment_filter("t1")
+
+    assert segment_filter(SimpleNamespace(absolute_uri="https://cdn.test/hls/ad1.ts")) is True
+    # a token-differing query string still matches on the path
+    assert segment_filter(SimpleNamespace(absolute_uri="https://cdn.test/hls/ad1.ts?token=2")) is True
+    assert segment_filter(SimpleNamespace(absolute_uri="https://cdn.test/hls/main0.ts")) is False
+    # fetched once, then cached per track id
+    assert len(responses.calls) == 1
+
+
+@responses.activate
+def test_client_filter_keeps_everything_when_the_server_has_no_route():
+    service = make_service(None, status=404)
+    segment_filter = service.remote_segment_filter("t1")
+
+    assert segment_filter(SimpleNamespace(absolute_uri="https://cdn.test/hls/ad1.ts")) is False

unshackle/core/api/handlers.py

@@ -2903,6 +2903,51 @@ def handle_proxy_license(
     return web.json_response({"license": base64.b64encode(license_response).decode("ascii")})
 
 
+async def session_segment_filter_handler(
+    data: Dict[str, Any], session_id: str, request: Optional[web.Request] = None
+) -> web.Response:
+    """Run the service's HLS ``OnSegmentFilter`` for one track and return the unwanted segment URIs.
+
+    The client fetches the same media playlist again. Only the URIs leave the server; the
+    filter and the data it reads stay on it.
+    """
+    session = await get_validated_session(session_id, request)
+
+    track_id = data.get("track_id")
+    if not track_id:
+        raise APIError(APIErrorCode.INVALID_INPUT, "Missing required parameter: track_id")
+
+    track = session.tracks.get(track_id)
+    if not track:
+        raise APIError(
+            APIErrorCode.TRACK_NOT_FOUND,
+            f"Track not found in session: {track_id}",
+            details={"track_id": track_id, "session_id": session_id},
+        )
+
+    segment_filter = getattr(track, "OnSegmentFilter", None)
+    if not callable(segment_filter) or not getattr(track, "url", None):
+        return web.json_response({"unwanted": None})
+
+    try:
+        import m3u8
+        import requests
+
+        response = session.service_instance.session.get(track.url)
+        if not getattr(response, "ok", True):
+            raise ValueError(f"playlist request failed: {response.status_code}")
+        if isinstance(response, requests.Response):
+            response.encoding = response.encoding or "utf-8"
+        playlist = m3u8.loads(response.text, uri=str(track.url))
+        unwanted = [segment.absolute_uri for segment in playlist.segments if segment_filter(segment)]
+    except (Exception, SystemExit):
+        # `from None`: a debug-mode traceback would otherwise carry the service's exception chain
+        log.exception(f"Error running the segment filter for track {sanitize_log(str(track_id))}")
+        raise APIError(APIErrorCode.SERVICE_ERROR, "Could not run the segment filter for this track") from None
+
+    return web.json_response({"unwanted": unwanted})
+
+
 async def session_license_handler(
     data: Dict[str, Any], session_id: str, request: Optional[web.Request] = None
 ) -> web.Response:

unshackle/core/api/routes.py

@@ -40,6 +40,7 @@ from unshackle.core.api.handlers import (
     session_license_handler,
     session_prompt_get_handler,
     session_prompt_post_handler,
+    session_segment_filter_handler,
     session_segments_handler,
     session_titles_handler,
     session_tracks_handler,
@@ -1675,6 +1676,57 @@ async def session_segments(request: web.Request) -> web.Response:
         )
 
 
+@api_handler
+async def session_segment_filter(request: web.Request) -> web.Response:
+    """
+    Get the unwanted HLS segments for one track.
+    ---
+    summary: Get segment filter
+    description: >-
+      Run the service's HLS segment filter on the server and return the segment URIs the
+      client must skip. The client fetches the same media playlist itself.
+    parameters:
+      - name: session_id
+        in: path
+        required: true
+        schema:
+          type: string
+    requestBody:
+      required: true
+      content:
+        application/json:
+          schema:
+            type: object
+            required:
+              - track_id
+            properties:
+              track_id:
+                type: string
+                description: Track ID to run the segment filter for
+    responses:
+      '200':
+        description: >-
+          Absolute URIs of the unwanted segments, or null when the track has no segment filter
+      '404':
+        description: Remote session or track not found
+    """
+    session_id = request.match_info["session_id"]
+    try:
+        data = await request.json()
+    except Exception as e:
+        return build_error_response(
+            APIError(APIErrorCode.INVALID_INPUT, "Invalid JSON request body", details={"error": str(e)}),
+            request.app.get("debug_api", False),
+        )
+    try:
+        return await session_segment_filter_handler(data, session_id, request)
+    except Exception as e:
+        log.exception("Error in session segment filter")
+        return handle_api_exception(
+            e, context={"operation": "session_segment_filter"}, debug_mode=request.app.get("debug_api", False)
+        )
+
+
 @api_handler
 async def session_license(request: web.Request) -> web.Response:
     """
@@ -1907,6 +1959,7 @@ ROUTES: list[tuple[str, str, Handler, bool]] = [
     ("GET", "/api/session/{session_id}/titles", session_titles, True),
     ("POST", "/api/session/{session_id}/tracks", session_tracks, True),
     ("POST", "/api/session/{session_id}/segments", session_segments, True),
+    ("POST", "/api/session/{session_id}/segment_filter", session_segment_filter, True),
     ("POST", "/api/session/{session_id}/license", session_license, True),
     ("GET", "/api/session/{session_id}/prompt", session_prompt_get, True),
     ("POST", "/api/session/{session_id}/prompt", session_prompt_submit, True),

unshackle/core/remote_service.py

@@ -15,7 +15,7 @@ from datetime import date as date_
 from enum import Enum
 from http.cookiejar import CookieJar
 from pathlib import Path
-from typing import Any, Dict, Optional, Union
+from typing import Any, Callable, Dict, Optional, Union
 
 import click
 import requests
@@ -81,7 +81,9 @@ class RemoteClient:
         session.headers[self.auth_headers[self._auth_header_index]] = self.api_key
         return True
 
-    def request(self, method: str, endpoint: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+    def request(
+        self, method: str, endpoint: str, data: Optional[Dict[str, Any]] = None, optional: bool = False
+    ) -> Dict[str, Any]:
         url = f"{self.server_url}{endpoint}"
         while True:
             try:
@@ -97,17 +99,27 @@ class RemoteClient:
             if resp.status_code == 401 and self.next_auth_header():
                 continue
             break
-        result = resp.json()
         if resp.status_code >= 400:
-            error_msg = redact_secrets(str(result.get("message", resp.text)), data)
-            error_code = result.get("error_code", "UNKNOWN")
+            try:
+                detail = resp.json()
+            except ValueError:
+                detail = {}
+            error_msg = redact_secrets(str(detail.get("message", resp.text)), data)
+            error_code = detail.get("error_code", "UNKNOWN")
+            if optional:
+                log.debug(f"Optional endpoint {endpoint} unavailable [{error_code}]: {error_msg}")
+                return {}
             log.error(f"Server error [{error_code}]: {error_msg}")
             raise SystemExit(1)
-        return result
+        return resp.json()
 
     def post(self, endpoint: str, data: Dict[str, Any]) -> Dict[str, Any]:
         return self.request("post", endpoint, data)
 
+    def post_optional(self, endpoint: str, data: Dict[str, Any]) -> Dict[str, Any]:
+        """POST to an endpoint that an older server may not have. Empty dict when the route is missing."""
+        return self.request("post", endpoint, data, optional=True)
+
     def get(self, endpoint: str) -> Dict[str, Any]:
         return self.request("get", endpoint)
 
@@ -633,6 +645,7 @@ class RemoteService:
         self._chapters_by_title: Dict[str, list] = {}
         self._session_id: Optional[str] = None
         self._server_cdm_type: str = "widevine"
+        self._segment_filters: Dict[str, tuple[set[str], set[str]]] = {}
 
         self._session = requests.Session()
         self._session.headers.update(config.headers)
@@ -848,11 +861,39 @@ class RemoteService:
             )
         self._server_cdm_type = result.get("server_cdm_type", "widevine")
 
+        for track in [*tracks.videos, *tracks.audio, *tracks.subtitles]:
+            if track.descriptor == Track.Descriptor.HLS:
+                track.OnSegmentFilter = self.remote_segment_filter(str(track.id))
+
         self._tracks_by_title[title_id] = tracks
         self._chapters_by_title[title_id] = result.get("chapters", [])
 
         return tracks
 
+    def remote_segment_filter(self, track_id: str) -> Callable[[Any], bool]:
+        """The server's ``OnSegmentFilter`` result for one HLS track, fetched on first use.
+
+        The service keeps its filter; the client gets only the segment URIs to skip.
+        A server without the route returns nothing, so the client keeps every segment.
+        """
+
+        def segment_filter(segment: Any) -> bool:
+            unwanted = self._segment_filters.get(track_id)
+            if unwanted is None:
+                result = self.client.post_optional(
+                    f"/api/session/{self._session_id}/segment_filter", {"track_id": track_id}
+                )
+                uris = [str(u) for u in (result.get("unwanted") or [])]
+                unwanted = (set(uris), {u.split("?")[0] for u in uris})
+                self._segment_filters[track_id] = unwanted
+            exact, paths = unwanted
+            if not exact:
+                return False
+            uri = str(getattr(segment, "absolute_uri", "") or "")
+            return uri in exact or uri.split("?")[0] in paths
+
+        return segment_filter
+
     def resolve_server_keys(self, title: Title_T) -> None:
         """Get the DRM content keys through the server CDM for all tracks on a title.