diff --git a/apps/backend/ARCHITECTURE.md b/apps/backend/ARCHITECTURE.md
index e17fab66..a77c0985 100644
--- a/apps/backend/ARCHITECTURE.md
+++ b/apps/backend/ARCHITECTURE.md
@@ -68,7 +68,7 @@ Responsibilities:
 Custom routes:
 
 - `POST /api/analysis-runs/estimate`
-- `POST /api/analysis-runs`
+- `POST /api/analysis-runs` — multipart upload OR URL ingestion. Provide *exactly one* of `track` (multipart `UploadFile`) or `url` (form field with a public `http`/`https` URL). URL mode is SSRF-guarded against private/loopback/link-local addresses and enforces the same 100 MiB cap via streaming. See [`url_ingest.py`](url_ingest.py).
 - `GET /api/analysis-runs/{run_id}`
 - `DELETE /api/analysis-runs/{run_id}`
 - `GET /api/analysis-runs/{run_id}/artifacts` and `…/artifacts/{artifact_id}`
diff --git a/apps/backend/server.py b/apps/backend/server.py
index 2e46cd39..53bc39aa 100644
--- a/apps/backend/server.py
+++ b/apps/backend/server.py
@@ -48,6 +48,7 @@
 from utils.cleanup import cleanup_artifacts
 import csv_export
 import upload_limits
+import url_ingest
 from server_upload import (  # noqa: F401 — re-exported for test backward compat
     LEGACY_ENDPOINT_SUNSET,
     ERROR_PHASE_LOCAL_DSP,
@@ -580,6 +581,80 @@ async def _create_analysis_run_record(
     return runtime, created["runId"]
 
 
+async def _create_analysis_run_record_from_url(
+    *,
+    url: str,
+    owner_user_id: str,
+    analysis_mode: str,
+    pitch_note_mode: str,
+    pitch_note_backend: str,
+    interpretation_mode: str,
+    interpretation_profile: str,
+    interpretation_model: str | None,
+    legacy_request_id: str | None = None,
+) -> tuple[AnalysisRuntime, str]:
+    """Parallel of ``_create_analysis_run_record`` but for URL-fetched audio.
+
+    The actual fetch is dispatched to a worker thread so the (blocking)
+    ``requests`` call doesn't pin the event loop. Typed URL-ingest
+    errors are surfaced unchanged to the caller, which is responsible
+    for translating them into HTTP envelopes.
+    """
+    runtime = get_analysis_runtime()
+    analysis_mode = _resolve_analysis_mode_value(analysis_mode)
+    if interpretation_mode != "off":
+        _resolve_interpretation_profile_config(interpretation_profile)
+
+    fetched = await asyncio.to_thread(url_ingest.fetch_url_to_bytes, url)
+
+    created = runtime.create_run(
+        filename=fetched.filename,
+        content=fetched.content,
+        mime_type=fetched.mime_type,
+        owner_user_id=owner_user_id,
+        analysis_mode=analysis_mode,
+        pitch_note_mode=pitch_note_mode,
+        pitch_note_backend=pitch_note_backend,
+        interpretation_mode=interpretation_mode,
+        interpretation_profile=interpretation_profile,
+        interpretation_model=interpretation_model,
+        legacy_request_id=legacy_request_id,
+    )
+    return runtime, created["runId"]
+
+
+def _url_ingest_error_response(exc: url_ingest.UrlIngestionError) -> JSONResponse:
+    """Map a typed URL-ingest error onto the canonical error envelope.
+
+    Status code and ``retryable`` flag follow the kind of failure:
+
+    - ``URL_INVALID`` / ``URL_BLOCKED_PRIVATE_HOST`` → 400, not
+      retryable (request will never succeed as-is).
+    - ``URL_TOO_LARGE`` → 413, not retryable (size is fundamental).
+    - ``URL_FETCH_FAILED`` → 502, retryable (transient upstream
+      problem; client can reasonably retry the same request).
+    """
+    code = exc.code
+    if code == "URL_TOO_LARGE":
+        status_code = 413
+        retryable = False
+    elif code == "URL_FETCH_FAILED":
+        status_code = 502
+        retryable = True
+    else:
+        status_code = 400
+        retryable = False
+    return JSONResponse(
+        status_code=status_code,
+        content={
+            "error": {
+                "code": code,
+                "message": str(exc),
+                "retryable": retryable,
+            }
+        },
+    )
+
 
 async def _estimate_analysis_run(
     *,
@@ -1819,7 +1894,8 @@ def _legacy_upload_too_large_file_response(request_id: str) -> JSONResponse:
 
 @app.post("/api/analysis-runs")
 async def create_analysis_run(
-    track: UploadFile = File(...),
+    track: UploadFile | None = File(None),
+    url: str | None = Form(None),
     analysis_mode: str = Form("full"),
     pitch_note_mode: str = Form("off"),
     pitch_note_backend: str = Form("auto"),
@@ -1829,35 +1905,113 @@ async def create_analysis_run(
     x_asa_user_id: str | None = Header(None),
     x_asa_user_email: str | None = Header(None),
 ) -> JSONResponse:
+    """Create an analysis run from either a multipart upload OR a public URL.
+
+    Exactly one of ``track`` (multipart) or ``url`` (form field) must be
+    provided. URL ingestion fetches the file server-side, subject to:
+
+    - HTTP/HTTPS only (no ``file://``, ``ftp://``, etc.).
+    - SSRF guard: hostnames that resolve to private, loopback, or
+      link-local IPs are rejected pre-flight.
+    - Size cap: same 100 MiB limit as multipart uploads
+      (``upload_limits.MAX_UPLOAD_SIZE_BYTES``).
+    - Timeouts: 30 s connect, 60 s read per chunk.
+
+    See :mod:`url_ingest` for the validation/fetch contract.
+    """
     invalid_backend_response = _pitch_note_backend_unsupported_response(pitch_note_backend)
     if invalid_backend_response is not None:
         return invalid_backend_response
 
-    upload_size = _uploaded_file_size_bytes(track)
-    if upload_size is not None and upload_size > upload_limits.MAX_UPLOAD_SIZE_BYTES:
-        return _canonical_upload_too_large_file_response()
+    # Exactly-one validation: caller must provide track XOR url. The
+    # distinction matters because both modes go through different
+    # validation paths.
+    #
+    # ``isinstance(url, str)`` is intentional rather than ``url is not None``:
+    # when this route is called directly in tests (not through HTTP
+    # dispatch), FastAPI's ``Form(None)`` default is an unresolved
+    # sentinel object — neither None nor a string. Existing multipart
+    # tests pre-date the URL field and don't pass ``url`` at all, so we
+    # must treat the sentinel as "not provided" rather than calling
+    # ``.strip()`` on it.
+    track_provided = track is not None and getattr(track, "filename", None)
+    url_provided = isinstance(url, str) and url.strip() != ""
+    if track_provided and url_provided:
+        if track is not None:
+            await track.close()
+        return JSONResponse(
+            status_code=400,
+            content={
+                "error": {
+                    "code": "AMBIGUOUS_AUDIO_SOURCE",
+                    "message": (
+                        "Provide exactly one of 'track' (multipart upload) or "
+                        "'url' (form field), not both."
+                    ),
+                    "retryable": False,
+                }
+            },
+        )
+    if not track_provided and not url_provided:
+        if track is not None:
+            await track.close()
+        return JSONResponse(
+            status_code=400,
+            content={
+                "error": {
+                    "code": "MISSING_AUDIO_SOURCE",
+                    "message": (
+                        "Provide either 'track' (multipart upload) or "
+                        "'url' (form field) as the audio source."
+                    ),
+                    "retryable": False,
+                }
+            },
+        )
+
+    # Pre-flight size check is only meaningful for multipart uploads;
+    # URL fetches enforce the limit during streaming inside url_ingest.
+    if track_provided:
+        upload_size = _uploaded_file_size_bytes(track)
+        if upload_size is not None and upload_size > upload_limits.MAX_UPLOAD_SIZE_BYTES:
+            return _canonical_upload_too_large_file_response()
 
     user_context = _resolve_route_user_context(x_asa_user_id, x_asa_user_email)
     if isinstance(user_context, JSONResponse):
-        await track.close()
+        if track is not None:
+            await track.close()
         return user_context
     try:
-        runtime, run_id = await _create_analysis_run_record(
-            track=track,
-            owner_user_id=user_context.user_id,
-            analysis_mode=analysis_mode,
-            pitch_note_mode=pitch_note_mode,
-            pitch_note_backend=pitch_note_backend,
-            interpretation_mode=interpretation_mode,
-            interpretation_profile=interpretation_profile,
-            interpretation_model=interpretation_model,
-        )
+        if url_provided:
+            runtime, run_id = await _create_analysis_run_record_from_url(
+                url=url,  # type: ignore[arg-type]
+                owner_user_id=user_context.user_id,
+                analysis_mode=analysis_mode,
+                pitch_note_mode=pitch_note_mode,
+                pitch_note_backend=pitch_note_backend,
+                interpretation_mode=interpretation_mode,
+                interpretation_profile=interpretation_profile,
+                interpretation_model=interpretation_model,
+            )
+        else:
+            runtime, run_id = await _create_analysis_run_record(
+                track=track,  # type: ignore[arg-type]
+                owner_user_id=user_context.user_id,
+                analysis_mode=analysis_mode,
+                pitch_note_mode=pitch_note_mode,
+                pitch_note_backend=pitch_note_backend,
+                interpretation_mode=interpretation_mode,
+                interpretation_profile=interpretation_profile,
+                interpretation_model=interpretation_model,
+            )
         return JSONResponse(
             content=_normalize_run_snapshot(
                 runtime.get_run(run_id, owner_user_id=user_context.user_id),
                 runtime,
             )
         )
+    except url_ingest.UrlIngestionError as exc:
+        return _url_ingest_error_response(exc)
     except UnsupportedPitchNoteBackendError as exc:
         return JSONResponse(
             status_code=400,
@@ -1889,7 +2043,8 @@ async def create_analysis_run(
             },
         )
     finally:
-        await track.close()
+        if track is not None:
+            await track.close()
 
 
 @app.post("/api/analysis-runs/estimate")
diff --git a/apps/backend/tests/test_server.py b/apps/backend/tests/test_server.py
index 676bc745..d917a43a 100644
--- a/apps/backend/tests/test_server.py
+++ b/apps/backend/tests/test_server.py
@@ -4204,6 +4204,230 @@ def test_aliases_scoped_per_device_not_global(self) -> None:
         self.assertEqual(self._codes(warnings), ["UNKNOWN_PARAMETER"])
 
 
+class CreateAnalysisRunUrlIngestionTests(unittest.TestCase):
+    """Route tests for POST /api/analysis-runs URL-mode.
+
+    The pure URL-fetch logic is covered by tests/test_url_ingest.py.
+    These tests verify the HTTP shell: source-selection branching,
+    error envelopes for each failure mode, and that a successful URL
+    fetch flows through the same downstream create_run path that a
+    multipart upload would.
+    """
+
+    def _decode_json_response(self, response) -> dict:
+        return json.loads(response.body.decode("utf-8"))
+
+    def test_url_mode_success_creates_a_run(self) -> None:
+        from analysis_runtime import AnalysisRuntime
+        import url_ingest
+
+        fetched = url_ingest.FetchedUrl(
+            content=b"fake-audio-from-url",
+            filename="track.mp3",
+            mime_type="audio/mpeg",
+        )
+
+        with tempfile.TemporaryDirectory(prefix="asa_url_ingest_") as temp_dir:
+            runtime = AnalysisRuntime(Path(temp_dir) / "runtime")
+            with patch.object(server, "get_analysis_runtime", return_value=runtime), \
+                 patch.object(
+                     server.url_ingest,
+                     "fetch_url_to_bytes",
+                     return_value=fetched,
+                 ):
+                response = asyncio.run(
+                    server.create_analysis_run(
+                        track=None,
+                        url="https://example.com/track.mp3",
+                        **self.DEFAULT_FORM_KWARGS,
+                    )
+                )
+
+        payload = self._decode_json_response(response)
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("runId", payload)
+        self.assertEqual(payload["stages"]["measurement"]["status"], "queued")
+
+    # Default form-field kwargs used by every test below. When calling the
+    # route function directly (not through HTTP dispatch), FastAPI's
+    # ``Form(...)`` defaults are unresolved sentinel objects, not the
+    # strings they wrap — so each test must pass the full set explicitly.
+    DEFAULT_FORM_KWARGS = dict(
+        analysis_mode="full",
+        pitch_note_mode="off",
+        pitch_note_backend="auto",
+        interpretation_mode="off",
+        interpretation_profile="producer_summary",
+        interpretation_model=None,
+    )
+
+    def test_neither_track_nor_url_returns_missing_audio_source(self) -> None:
+        from analysis_runtime import AnalysisRuntime
+
+        with tempfile.TemporaryDirectory(prefix="asa_url_ingest_") as temp_dir:
+            runtime = AnalysisRuntime(Path(temp_dir) / "runtime")
+            with patch.object(server, "get_analysis_runtime", return_value=runtime):
+                response = asyncio.run(
+                    server.create_analysis_run(
+                        track=None,
+                        url=None,
+                        **self.DEFAULT_FORM_KWARGS,
+                    )
+                )
+
+        self.assertEqual(response.status_code, 400)
+        payload = self._decode_json_response(response)
+        self.assertEqual(payload["error"]["code"], "MISSING_AUDIO_SOURCE")
+
+    def test_both_track_and_url_returns_ambiguous_audio_source(self) -> None:
+        from analysis_runtime import AnalysisRuntime
+        from fastapi import UploadFile
+
+        with tempfile.TemporaryDirectory(prefix="asa_url_ingest_") as temp_dir:
+            runtime = AnalysisRuntime(Path(temp_dir) / "runtime")
+            track = UploadFile(filename="track.mp3", file=io.BytesIO(b"x"))
+            with patch.object(server, "get_analysis_runtime", return_value=runtime):
+                response = asyncio.run(
+                    server.create_analysis_run(
+                        track=track,
+                        url="https://example.com/track.mp3",
+                        **self.DEFAULT_FORM_KWARGS,
+                    )
+                )
+
+        self.assertEqual(response.status_code, 400)
+        payload = self._decode_json_response(response)
+        self.assertEqual(payload["error"]["code"], "AMBIGUOUS_AUDIO_SOURCE")
+
+    def test_url_invalid_returns_400_with_url_invalid(self) -> None:
+        from analysis_runtime import AnalysisRuntime
+        import url_ingest
+
+        with tempfile.TemporaryDirectory(prefix="asa_url_ingest_") as temp_dir:
+            runtime = AnalysisRuntime(Path(temp_dir) / "runtime")
+            with patch.object(server, "get_analysis_runtime", return_value=runtime), \
+                 patch.object(
+                     server.url_ingest,
+                     "fetch_url_to_bytes",
+                     side_effect=url_ingest.UrlInvalidError("bad scheme"),
+                 ):
+                response = asyncio.run(
+                    server.create_analysis_run(
+                        track=None,
+                        url="file:///etc/passwd",
+                        **self.DEFAULT_FORM_KWARGS,
+                    )
+                )
+
+        self.assertEqual(response.status_code, 400)
+        payload = self._decode_json_response(response)
+        self.assertEqual(payload["error"]["code"], "URL_INVALID")
+
+    def test_url_blocked_private_host_returns_400(self) -> None:
+        from analysis_runtime import AnalysisRuntime
+        import url_ingest
+
+        with tempfile.TemporaryDirectory(prefix="asa_url_ingest_") as temp_dir:
+            runtime = AnalysisRuntime(Path(temp_dir) / "runtime")
+            with patch.object(server, "get_analysis_runtime", return_value=runtime), \
+                 patch.object(
+                     server.url_ingest,
+                     "fetch_url_to_bytes",
+                     side_effect=url_ingest.UrlBlockedPrivateHostError(
+                         "10.0.0.5 is private"
+                     ),
+                 ):
+                response = asyncio.run(
+                    server.create_analysis_run(
+                        track=None,
+                        url="http://internal.example.com/audio.mp3",
+                        **self.DEFAULT_FORM_KWARGS,
+                    )
+                )
+
+        self.assertEqual(response.status_code, 400)
+        payload = self._decode_json_response(response)
+        self.assertEqual(payload["error"]["code"], "URL_BLOCKED_PRIVATE_HOST")
+
+    def test_url_too_large_returns_413(self) -> None:
+        from analysis_runtime import AnalysisRuntime
+        import url_ingest
+
+        with tempfile.TemporaryDirectory(prefix="asa_url_ingest_") as temp_dir:
+            runtime = AnalysisRuntime(Path(temp_dir) / "runtime")
+            with patch.object(server, "get_analysis_runtime", return_value=runtime), \
+                 patch.object(
+                     server.url_ingest,
+                     "fetch_url_to_bytes",
+                     side_effect=url_ingest.UrlTooLargeError("body 200 MiB"),
+                 ):
+                response = asyncio.run(
+                    server.create_analysis_run(
+                        track=None,
+                        url="https://example.com/huge.flac",
+                        **self.DEFAULT_FORM_KWARGS,
+                    )
+                )
+
+        self.assertEqual(response.status_code, 413)
+        payload = self._decode_json_response(response)
+        self.assertEqual(payload["error"]["code"], "URL_TOO_LARGE")
+
+    def test_url_fetch_failed_returns_502(self) -> None:
+        from analysis_runtime import AnalysisRuntime
+        import url_ingest
+
+        with tempfile.TemporaryDirectory(prefix="asa_url_ingest_") as temp_dir:
+            runtime = AnalysisRuntime(Path(temp_dir) / "runtime")
+            with patch.object(server, "get_analysis_runtime", return_value=runtime), \
+                 patch.object(
+                     server.url_ingest,
+                     "fetch_url_to_bytes",
+                     side_effect=url_ingest.UrlFetchFailedError("timed out"),
+                 ):
+                response = asyncio.run(
+                    server.create_analysis_run(
+                        track=None,
+                        url="https://example.com/slow.mp3",
+                        **self.DEFAULT_FORM_KWARGS,
+                    )
+                )
+
+        self.assertEqual(response.status_code, 502)
+        payload = self._decode_json_response(response)
+        self.assertEqual(payload["error"]["code"], "URL_FETCH_FAILED")
+        # 502 means upstream issue — the client can reasonably retry.
+        # ARCHITECTURE.md contract says envelopes always carry retryable;
+        # this is the first 502 the route returns, so it's the cleanest
+        # place to anchor the assertion.
+        self.assertTrue(payload["error"]["retryable"])
+
+    def test_url_invalid_envelope_is_not_retryable(self) -> None:
+        from analysis_runtime import AnalysisRuntime
+        import url_ingest
+
+        with tempfile.TemporaryDirectory(prefix="asa_url_ingest_") as temp_dir:
+            runtime = AnalysisRuntime(Path(temp_dir) / "runtime")
+            with patch.object(server, "get_analysis_runtime", return_value=runtime), \
+                 patch.object(
+                     server.url_ingest,
+                     "fetch_url_to_bytes",
+                     side_effect=url_ingest.UrlInvalidError("bad scheme"),
+                 ):
+                response = asyncio.run(
+                    server.create_analysis_run(
+                        track=None,
+                        url="file:///etc/passwd",
+                        **self.DEFAULT_FORM_KWARGS,
+                    )
+                )
+
+        payload = self._decode_json_response(response)
+        # Malformed URL won't succeed on retry without a different URL,
+        # so retryable must be False.
+        self.assertFalse(payload["error"]["retryable"])
+
+
 class CsvExportRouteTests(unittest.TestCase):
     """Route tests for GET /api/analysis-runs/{run_id}/export/csv/{field_path}.
 
diff --git a/apps/backend/tests/test_url_ingest.py b/apps/backend/tests/test_url_ingest.py
new file mode 100644
index 00000000..5661f24e
--- /dev/null
+++ b/apps/backend/tests/test_url_ingest.py
@@ -0,0 +1,395 @@
+"""Unit tests for url_ingest — URL validation, SSRF guard, fetch error mapping.
+
+The route-level tests in test_server.py exercise the HTTP shell;
+these tests focus on the pure-logic surface of url_ingest.py: given
+specific URL shapes and mock HTTP responses, does the right error
+class come back, does the size limit get enforced, does the SSRF
+guard block private addresses?
+"""
+
+import sys
+import unittest
+from pathlib import Path
+from unittest import mock
+
+
+_BACKEND_ROOT = Path(__file__).resolve().parents[1]
+if str(_BACKEND_ROOT) not in sys.path:
+    sys.path.insert(0, str(_BACKEND_ROOT))
+
+
+import url_ingest  # noqa: E402 — load after sys.path is set
+
+
+def _make_addrinfo_for(ip: str):
+    """Build a getaddrinfo-style result for one IPv4 address."""
+    import socket as _socket
+
+    family = _socket.AF_INET6 if ":" in ip else _socket.AF_INET
+    sockaddr = (ip, 0, 0, 0) if family == _socket.AF_INET6 else (ip, 0)
+    return [(family, _socket.SOCK_STREAM, 0, "", sockaddr)]
+
+
+def _make_mock_response(
+    *,
+    status: int = 200,
+    body: bytes = b"audio-bytes",
+    content_type: str = "audio/mpeg",
+    content_length: str | None = None,
+    chunk_size: int = 4096,
+) -> mock.MagicMock:
+    """A stand-in for the requests.Response that ``requests.get`` returns
+    as a context manager. Streams ``body`` in chunks of ``chunk_size``."""
+    headers = {"Content-Type": content_type}
+    if content_length is not None:
+        headers["Content-Length"] = content_length
+
+    def iter_content(chunk_size=4096):  # noqa: ARG001 — match requests API
+        # Yield body in fixed-size pieces.
+        for i in range(0, len(body), chunk_size):
+            yield body[i : i + chunk_size]
+
+    response = mock.MagicMock(
+        name="response",
+        status_code=status,
+        headers=headers,
+    )
+    response.iter_content.side_effect = iter_content
+    # Make the response a usable context manager: __enter__ returns the
+    # response itself, __exit__ does nothing.
+    response.__enter__ = mock.MagicMock(return_value=response)
+    response.__exit__ = mock.MagicMock(return_value=False)
+    return response
+
+
+class UrlShapeValidationTests(unittest.TestCase):
+    """The cheap, deterministic URL-string checks."""
+
+    def test_empty_string_rejected(self):
+        with self.assertRaises(url_ingest.UrlInvalidError):
+            url_ingest._validate_url_shape("")
+
+    def test_too_long_rejected(self):
+        too_long = "https://example.com/" + ("a" * (url_ingest.MAX_URL_LENGTH + 1))
+        with self.assertRaises(url_ingest.UrlInvalidError):
+            url_ingest._validate_url_shape(too_long)
+
+    def test_file_scheme_rejected(self):
+        with self.assertRaises(url_ingest.UrlInvalidError) as ctx:
+            url_ingest._validate_url_shape("file:///etc/passwd")
+        self.assertIn("scheme", str(ctx.exception).lower())
+
+    def test_ftp_scheme_rejected(self):
+        with self.assertRaises(url_ingest.UrlInvalidError):
+            url_ingest._validate_url_shape("ftp://example.com/audio.mp3")
+
+    def test_no_hostname_rejected(self):
+        with self.assertRaises(url_ingest.UrlInvalidError):
+            url_ingest._validate_url_shape("https:///audio.mp3")
+
+    def test_http_accepted(self):
+        # No exception
+        url_ingest._validate_url_shape("http://example.com/audio.mp3")
+
+    def test_https_accepted(self):
+        url_ingest._validate_url_shape("https://example.com/audio.mp3")
+
+
+class SsrfGuardTests(unittest.TestCase):
+    """Hostnames that resolve to private/loopback/etc. must be rejected
+    before any HTTP request is dispatched."""
+
+    def _assert_blocks(self, ip: str) -> None:
+        with mock.patch(
+            "socket.getaddrinfo", return_value=_make_addrinfo_for(ip)
+        ):
+            with self.assertRaises(url_ingest.UrlBlockedPrivateHostError):
+                url_ingest._assert_host_is_public("blocked.example.com")
+
+    def _assert_allows(self, ip: str) -> None:
+        with mock.patch(
+            "socket.getaddrinfo", return_value=_make_addrinfo_for(ip)
+        ):
+            # Should not raise.
+            url_ingest._assert_host_is_public("public.example.com")
+
+    def test_blocks_ipv4_loopback(self):
+        self._assert_blocks("127.0.0.1")
+
+    def test_blocks_ipv4_rfc1918_10(self):
+        self._assert_blocks("10.0.0.5")
+
+    def test_blocks_ipv4_rfc1918_192(self):
+        self._assert_blocks("192.168.1.50")
+
+    def test_blocks_ipv4_rfc1918_172(self):
+        self._assert_blocks("172.20.0.1")
+
+    def test_blocks_ipv4_link_local(self):
+        self._assert_blocks("169.254.169.254")  # AWS metadata endpoint
+
+    def test_blocks_ipv6_loopback(self):
+        self._assert_blocks("::1")
+
+    def test_blocks_cgnat_rfc6598(self):
+        # 100.64.0.0/10 is CGNAT. Python's ipaddress.is_private does not
+        # return True for this range, so the helper must catch it
+        # explicitly. Sample a few addresses inside the block.
+        self._assert_blocks("100.64.0.1")
+        self._assert_blocks("100.100.50.50")
+        self._assert_blocks("100.127.255.254")
+
+    def test_allows_addresses_just_outside_cgnat(self):
+        # 100.63.x.x and 100.128.x.x are outside the /10 range and
+        # should pass — verifies the netmask is correct, not just
+        # a substring match on "100.".
+        self._assert_allows("100.63.255.255")
+        self._assert_allows("100.128.0.1")
+
+    def test_allows_public_ipv4(self):
+        self._assert_allows("8.8.8.8")  # Google DNS
+
+    def test_allows_public_ipv6(self):
+        self._assert_allows("2001:4860:4860::8888")  # Google DNS v6
+
+    def test_dns_failure_surfaces_as_fetch_failed(self):
+        import socket as _socket
+
+        with mock.patch(
+            "socket.getaddrinfo", side_effect=_socket.gaierror("no such host")
+        ):
+            with self.assertRaises(url_ingest.UrlFetchFailedError):
+                url_ingest._assert_host_is_public("nx.example.com")
+
+
+class FilenameExtractionTests(unittest.TestCase):
+    """Filename comes from the last URL path segment with a safe fallback."""
+
+    def test_simple_path(self):
+        self.assertEqual(url_ingest._extract_filename("/audio.mp3"), "audio.mp3")
+
+    def test_nested_path(self):
+        self.assertEqual(
+            url_ingest._extract_filename("/uploads/tracks/song.flac"),
+            "song.flac",
+        )
+
+    def test_trailing_slash(self):
+        self.assertEqual(url_ingest._extract_filename("/audio.mp3/"), "audio.mp3")
+
+    def test_empty_path_falls_back(self):
+        self.assertEqual(url_ingest._extract_filename(""), "url_audio.bin")
+
+    def test_root_falls_back(self):
+        self.assertEqual(url_ingest._extract_filename("/"), "url_audio.bin")
+
+    def test_dotdot_falls_back(self):
+        # Defense in depth — we never use this as a filesystem path,
+        # but the helper shouldn't return "..".
+        self.assertEqual(url_ingest._extract_filename("/../audio.mp3"), "audio.mp3")
+        self.assertEqual(url_ingest._extract_filename("/.."), "url_audio.bin")
+
+    def test_url_encoded_filename_is_decoded(self):
+        self.assertEqual(
+            url_ingest._extract_filename("/track%20one.mp3"),
+            "track one.mp3",
+        )
+
+
+class MimeTypePickerTests(unittest.TestCase):
+    """Content-Type wins; filename guess is the fallback."""
+
+    def test_content_type_header_wins(self):
+        self.assertEqual(
+            url_ingest._pick_mime_type("audio/mpeg", "/anything.bin"),
+            "audio/mpeg",
+        )
+
+    def test_content_type_with_charset_is_stripped(self):
+        self.assertEqual(
+            url_ingest._pick_mime_type(
+                "audio/mpeg; charset=binary", "/x.bin"
+            ),
+            "audio/mpeg",
+        )
+
+    def test_octet_stream_triggers_filename_fallback(self):
+        # application/octet-stream is too generic — fall through.
+        self.assertEqual(
+            url_ingest._pick_mime_type(
+                "application/octet-stream", "/audio.flac"
+            ),
+            "audio/flac",
+        )
+
+    def test_no_content_type_falls_back_to_filename(self):
+        self.assertEqual(
+            url_ingest._pick_mime_type("", "/audio.wav"), "audio/x-wav"
+        )
+
+    def test_unknown_extension_falls_back_to_octet_stream(self):
+        # ``.qqqzz`` is not in any standard mimetypes database, so the
+        # fallback path runs.
+        self.assertEqual(
+            url_ingest._pick_mime_type("", "/audio.qqqzz"),
+            "application/octet-stream",
+        )
+
+
+class FetchUrlToBytesTests(unittest.TestCase):
+    """End-to-end fetch path with mocked requests.get."""
+
+    PUBLIC_URL = "https://example.com/track.mp3"
+
+    def _patch_dns_and_get(self, response, public_ip="93.184.216.34"):
+        """Stack the patches needed for a happy-path fetch."""
+        return mock.patch.multiple(
+            "url_ingest",
+            requests=mock.DEFAULT,
+        ), mock.patch(
+            "socket.getaddrinfo",
+            return_value=_make_addrinfo_for(public_ip),
+        )
+
+    def test_happy_path_returns_content_filename_mime(self):
+        response = _make_mock_response(
+            status=200,
+            body=b"abc" * 1000,  # 3000 bytes
+            content_type="audio/mpeg",
+        )
+        with mock.patch(
+            "socket.getaddrinfo",
+            return_value=_make_addrinfo_for("93.184.216.34"),
+        ), mock.patch.object(url_ingest.requests, "get", return_value=response):
+            fetched = url_ingest.fetch_url_to_bytes(self.PUBLIC_URL)
+
+        self.assertEqual(fetched.content, b"abc" * 1000)
+        self.assertEqual(fetched.filename, "track.mp3")
+        self.assertEqual(fetched.mime_type, "audio/mpeg")
+
+    def test_http_error_status_raises_fetch_failed(self):
+        response = _make_mock_response(status=404, body=b"not found")
+        with mock.patch(
+            "socket.getaddrinfo",
+            return_value=_make_addrinfo_for("93.184.216.34"),
+        ), mock.patch.object(url_ingest.requests, "get", return_value=response):
+            with self.assertRaises(url_ingest.UrlFetchFailedError) as ctx:
+                url_ingest.fetch_url_to_bytes(self.PUBLIC_URL)
+            self.assertIn("404", str(ctx.exception))
+
+    def test_redirect_is_rejected_not_followed(self):
+        # An attacker who controls a public host could 302 us to a
+        # private/loopback target (e.g. AWS metadata at 169.254.169.254)
+        # to bypass the SSRF guard. We refuse to follow redirects and
+        # surface 3xx as a fetch failure.
+        response = _make_mock_response(status=302, body=b"")
+        # Add a Location header that points to a sensitive private target
+        # — the test asserts we don't even *try* to fetch it.
+        response.headers["Location"] = "http://169.254.169.254/latest/meta-data/"
+        with mock.patch(
+            "socket.getaddrinfo",
+            return_value=_make_addrinfo_for("93.184.216.34"),
+        ), mock.patch.object(
+            url_ingest.requests, "get", return_value=response
+        ) as mock_get:
+            with self.assertRaises(url_ingest.UrlFetchFailedError) as ctx:
+                url_ingest.fetch_url_to_bytes(self.PUBLIC_URL)
+            self.assertIn("redirect", str(ctx.exception).lower())
+            # requests.get must have been called with allow_redirects=False
+            # — that's the contract the SSRF defense rests on.
+            self.assertEqual(
+                mock_get.call_args.kwargs["allow_redirects"],
+                False,
+            )
+
+    def test_redirect_301_also_rejected(self):
+        response = _make_mock_response(status=301, body=b"")
+        response.headers["Location"] = "https://elsewhere.example.com/audio.mp3"
+        with mock.patch(
+            "socket.getaddrinfo",
+            return_value=_make_addrinfo_for("93.184.216.34"),
+        ), mock.patch.object(url_ingest.requests, "get", return_value=response):
+            with self.assertRaises(url_ingest.UrlFetchFailedError):
+                url_ingest.fetch_url_to_bytes(self.PUBLIC_URL)
+
+    def test_content_length_too_large_raises(self):
+        response = _make_mock_response(
+            status=200,
+            body=b"x",
+            # Declare a 1 GiB body — exceeds default 100 MiB limit.
+            content_length=str(1024 * 1024 * 1024),
+        )
+        with mock.patch(
+            "socket.getaddrinfo",
+            return_value=_make_addrinfo_for("93.184.216.34"),
+        ), mock.patch.object(url_ingest.requests, "get", return_value=response):
+            with self.assertRaises(url_ingest.UrlTooLargeError):
+                url_ingest.fetch_url_to_bytes(self.PUBLIC_URL)
+
+    def test_streaming_overrun_raises(self):
+        # No Content-Length set; body is larger than max_bytes override.
+        response = _make_mock_response(
+            status=200,
+            body=b"x" * 5_000,
+            content_type="audio/mpeg",
+        )
+        with mock.patch(
+            "socket.getaddrinfo",
+            return_value=_make_addrinfo_for("93.184.216.34"),
+        ), mock.patch.object(url_ingest.requests, "get", return_value=response):
+            with self.assertRaises(url_ingest.UrlTooLargeError):
+                url_ingest.fetch_url_to_bytes(self.PUBLIC_URL, max_bytes=1_000)
+
+    def test_empty_body_raises_fetch_failed(self):
+        response = _make_mock_response(status=200, body=b"")
+        with mock.patch(
+            "socket.getaddrinfo",
+            return_value=_make_addrinfo_for("93.184.216.34"),
+        ), mock.patch.object(url_ingest.requests, "get", return_value=response):
+            with self.assertRaises(url_ingest.UrlFetchFailedError) as ctx:
+                url_ingest.fetch_url_to_bytes(self.PUBLIC_URL)
+            self.assertIn("empty", str(ctx.exception).lower())
+
+    def test_private_host_blocks_before_http(self):
+        # If SSRF guard fires, we never reach requests.get.
+        mock_get = mock.MagicMock()
+        with mock.patch(
+            "socket.getaddrinfo",
+            return_value=_make_addrinfo_for("10.0.0.5"),
+        ), mock.patch.object(url_ingest.requests, "get", mock_get):
+            with self.assertRaises(url_ingest.UrlBlockedPrivateHostError):
+                url_ingest.fetch_url_to_bytes(self.PUBLIC_URL)
+            mock_get.assert_not_called()
+
+    def test_timeout_raises_fetch_failed(self):
+        import requests as _requests
+
+        with mock.patch(
+            "socket.getaddrinfo",
+            return_value=_make_addrinfo_for("93.184.216.34"),
+        ), mock.patch.object(
+            url_ingest.requests,
+            "get",
+            side_effect=_requests.exceptions.Timeout("read timeout"),
+        ):
+            with self.assertRaises(url_ingest.UrlFetchFailedError) as ctx:
+                url_ingest.fetch_url_to_bytes(self.PUBLIC_URL)
+            self.assertIn("timed out", str(ctx.exception).lower())
+
+    def test_connection_error_raises_fetch_failed(self):
+        import requests as _requests
+
+        with mock.patch(
+            "socket.getaddrinfo",
+            return_value=_make_addrinfo_for("93.184.216.34"),
+        ), mock.patch.object(
+            url_ingest.requests,
+            "get",
+            side_effect=_requests.exceptions.ConnectionError("refused"),
+        ):
+            with self.assertRaises(url_ingest.UrlFetchFailedError):
+                url_ingest.fetch_url_to_bytes(self.PUBLIC_URL)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/apps/backend/url_ingest.py b/apps/backend/url_ingest.py
new file mode 100644
index 00000000..03e429ee
--- /dev/null
+++ b/apps/backend/url_ingest.py
@@ -0,0 +1,390 @@
+"""URL-based ingestion for ``POST /api/analysis-runs``.
+
+Lets a producer point ASA at a publicly hosted audio file instead of
+uploading bytes through the browser. The server fetches, validates,
+and feeds the bytes into the same downstream pipeline as a multipart
+upload would.
+
+Scope (v1):
+
+- HTTP and HTTPS only. ``file://``, ``ftp://`` etc. are rejected.
+- SSRF guarded: hostnames that resolve to private, loopback, link-local,
+  or otherwise reserved IP space are rejected before any HTTP request
+  is made. This is a "first-line" guard — there is still a small DNS
+  TOCTOU window between resolution and fetch, but the alternative
+  (custom DNS resolver that pins the result through the HTTP client)
+  is out of scope for v1.
+- Streaming size enforcement: ``Content-Length`` is checked up-front
+  if present, and the body is also streamed-and-accumulated with an
+  abort if bytes exceed :data:`upload_limits.MAX_UPLOAD_SIZE_BYTES`.
+- Timeouts: connect + read timeout default 60 s; total wall clock
+  capped at 300 s.
+- The filename is extracted from the URL path; falls back to
+  ``url_audio.bin`` when missing. MIME type prefers the response
+  ``Content-Type`` header, then guesses from the filename.
+
+This module is pure I/O + validation. The HTTP route handler in
+``server.py`` is responsible for translating typed exceptions raised
+here into the canonical error envelope codes.
+"""
+
+from __future__ import annotations
+
+import ipaddress
+import socket
+from dataclasses import dataclass
+from typing import Final
+from urllib.parse import unquote, urlparse
+
+import requests
+
+import upload_limits
+
+
+__all__ = [
+    "FetchedUrl",
+    "UrlIngestionError",
+    "UrlInvalidError",
+    "UrlBlockedPrivateHostError",
+    "UrlFetchFailedError",
+    "UrlTooLargeError",
+    "fetch_url_to_bytes",
+]
+
+
+# Public defaults — keep them in sync with the route-handler docs.
+DEFAULT_CONNECT_TIMEOUT_S: Final[float] = 30.0
+DEFAULT_READ_TIMEOUT_S: Final[float] = 60.0
+DEFAULT_TOTAL_TIMEOUT_S: Final[float] = 300.0
+MAX_URL_LENGTH: Final[int] = 4096
+ALLOWED_SCHEMES: Final[frozenset[str]] = frozenset({"http", "https"})
+# Stream chunk size. Small enough to detect oversize quickly; large enough
+# that we're not blocked on syscall overhead.
+_FETCH_CHUNK_BYTES: Final[int] = 64 * 1024
+
+
+@dataclass(frozen=True)
+class FetchedUrl:
+    """The output of a successful URL fetch.
+
+    ``content`` is the full audio body in memory. The byte limit
+    (``MAX_UPLOAD_SIZE_BYTES``, 100 MiB) means this is bounded; we do
+    not stream to disk because the downstream ``create_run`` API
+    expects ``bytes`` and writes to its own artifact store from there.
+    """
+
+    content: bytes
+    filename: str
+    mime_type: str
+
+
+# ----------------------------------------------------------------------
+# Exception hierarchy
+# ----------------------------------------------------------------------
+
+
+class UrlIngestionError(Exception):
+    """Base class for URL-ingestion errors.
+
+    Subclasses map 1:1 to public error codes returned by the route
+    handler. The ``message`` is safe to surface to the client; do not
+    embed secrets or internal hostnames.
+    """
+
+    code: str = "URL_INGESTION_ERROR"
+
+
+class UrlInvalidError(UrlIngestionError):
+    """The URL is malformed, too long, or uses a disallowed scheme."""
+
+    code = "URL_INVALID"
+
+
+class UrlBlockedPrivateHostError(UrlIngestionError):
+    """The URL resolves to a private/loopback/link-local IP — SSRF guard.
+
+    Triggered before any HTTP request is dispatched. The TOCTOU window
+    between resolution-time and fetch-time is acknowledged as out of
+    scope for v1; a pinning DNS resolver is the v2 improvement.
+    """
+
+    code = "URL_BLOCKED_PRIVATE_HOST"
+
+
+class UrlFetchFailedError(UrlIngestionError):
+    """Network, DNS, timeout, or non-2xx response. Generic 'fetch failed'.
+
+    The message includes the underlying error class name (without
+    private internals) so operators can diagnose. Do not log the URL
+    itself at INFO level — treat as user input.
+    """
+
+    code = "URL_FETCH_FAILED"
+
+
+class UrlTooLargeError(UrlIngestionError):
+    """The response exceeds :data:`upload_limits.MAX_UPLOAD_SIZE_BYTES`.
+
+    Raised either from a declared ``Content-Length`` header or from
+    streaming the body and accumulating past the limit.
+    """
+
+    code = "URL_TOO_LARGE"
+
+
+# ----------------------------------------------------------------------
+# Public entry point
+# ----------------------------------------------------------------------
+
+
+def fetch_url_to_bytes(
+    url: str,
+    *,
+    max_bytes: int = upload_limits.MAX_UPLOAD_SIZE_BYTES,
+    connect_timeout_s: float = DEFAULT_CONNECT_TIMEOUT_S,
+    read_timeout_s: float = DEFAULT_READ_TIMEOUT_S,
+    user_agent: str = "AbletonSonicAnalyzer/url-ingest",
+) -> FetchedUrl:
+    """Validate ``url``, fetch its bytes, return content + filename + MIME.
+
+    Raises:
+        UrlInvalidError: malformed URL, oversize URL, or non-http(s) scheme.
+        UrlBlockedPrivateHostError: hostname resolves to private IP space.
+        UrlFetchFailedError: any network, DNS, or non-2xx failure.
+        UrlTooLargeError: response body exceeds ``max_bytes``.
+    """
+    _validate_url_shape(url)
+    parsed = urlparse(url)
+    _assert_host_is_public(parsed.hostname or "")
+
+    try:
+        with requests.get(
+            url,
+            stream=True,
+            timeout=(connect_timeout_s, read_timeout_s),
+            # Redirects are NOT followed automatically. The SSRF guard
+            # above only validates the initial hostname; if we followed
+            # a 3xx, an attacker controlling any public host could
+            # redirect us to a private/loopback target (e.g.
+            # 169.254.169.254 metadata endpoint) and bypass the guard.
+            # A 3xx is surfaced as a fetch failure; users should provide
+            # the canonical direct URL.
+            allow_redirects=False,
+            headers={"User-Agent": user_agent},
+        ) as response:
+            if 300 <= response.status_code < 400:
+                raise UrlFetchFailedError(
+                    f"Upstream returned HTTP {response.status_code} "
+                    f"(redirect). URL ingestion does not follow redirects; "
+                    f"provide a direct URL to the audio file."
+                )
+            if response.status_code >= 400:
+                raise UrlFetchFailedError(
+                    f"Upstream returned HTTP {response.status_code} for the URL."
+                )
+
+            # Up-front check on declared size before streaming any body.
+            declared_size = _parse_content_length(
+                response.headers.get("Content-Length")
+            )
+            if declared_size is not None and declared_size > max_bytes:
+                raise UrlTooLargeError(
+                    f"Upstream declared Content-Length {declared_size} bytes; "
+                    f"limit is {max_bytes} bytes."
+                )
+
+            chunks: list[bytes] = []
+            total_bytes = 0
+            for chunk in response.iter_content(chunk_size=_FETCH_CHUNK_BYTES):
+                if not chunk:
+                    continue
+                total_bytes += len(chunk)
+                if total_bytes > max_bytes:
+                    raise UrlTooLargeError(
+                        f"Response body exceeded {max_bytes} bytes; aborted."
+                    )
+                chunks.append(chunk)
+
+            if total_bytes == 0:
+                raise UrlFetchFailedError(
+                    "Upstream returned an empty body."
+                )
+
+            content = b"".join(chunks)
+            content_type = response.headers.get("Content-Type") or ""
+            mime_type = _pick_mime_type(content_type, parsed.path)
+            filename = _extract_filename(parsed.path)
+
+            return FetchedUrl(
+                content=content,
+                filename=filename,
+                mime_type=mime_type,
+            )
+
+    except UrlIngestionError:
+        raise
+    except requests.exceptions.Timeout as exc:
+        raise UrlFetchFailedError(
+            f"Timed out fetching URL: {type(exc).__name__}."
+        ) from exc
+    except requests.exceptions.ConnectionError as exc:
+        raise UrlFetchFailedError(
+            f"Connection error fetching URL: {type(exc).__name__}."
+        ) from exc
+    except requests.exceptions.RequestException as exc:
+        raise UrlFetchFailedError(
+            f"HTTP client error fetching URL: {type(exc).__name__}."
+        ) from exc
+
+
+# ----------------------------------------------------------------------
+# Validation
+# ----------------------------------------------------------------------
+
+
+def _validate_url_shape(url: str) -> None:
+    """Cheap, deterministic URL-string checks. No network."""
+    if not isinstance(url, str) or not url:
+        raise UrlInvalidError("URL must be a non-empty string.")
+    if len(url) > MAX_URL_LENGTH:
+        raise UrlInvalidError(
+            f"URL exceeds maximum length of {MAX_URL_LENGTH} characters."
+        )
+    try:
+        parsed = urlparse(url)
+    except Exception as exc:
+        raise UrlInvalidError(f"URL is not parseable: {type(exc).__name__}.") from exc
+    if parsed.scheme not in ALLOWED_SCHEMES:
+        raise UrlInvalidError(
+            f"URL scheme '{parsed.scheme}' is not allowed. "
+            f"Supported schemes: {', '.join(sorted(ALLOWED_SCHEMES))}."
+        )
+    if not parsed.hostname:
+        raise UrlInvalidError("URL must include a hostname.")
+
+
+def _assert_host_is_public(hostname: str) -> None:
+    """Reject hostnames that resolve to non-public IP space.
+
+    Performs the DNS resolution here so the SSRF check is done before
+    we send any HTTP request. A small TOCTOU window remains between
+    this resolution and the eventual fetch; the v2 fix is a custom
+    DNS resolver that pins the result through the HTTP client.
+    """
+    if not hostname:
+        raise UrlInvalidError("URL must include a hostname.")
+
+    # Resolve all A/AAAA records, not just the first; reject if *any*
+    # resolved address is non-public.
+    try:
+        addrinfo = socket.getaddrinfo(hostname, None)
+    except socket.gaierror as exc:
+        raise UrlFetchFailedError(
+            f"Could not resolve hostname: {type(exc).__name__}."
+        ) from exc
+
+    seen: set[str] = set()
+    for family, _socktype, _proto, _canon, sockaddr in addrinfo:
+        if family not in (socket.AF_INET, socket.AF_INET6):
+            continue
+        # IPv4 sockaddr = (ip, port); IPv6 = (ip, port, flow, scope).
+        ip_str = sockaddr[0]
+        if ip_str in seen:
+            continue
+        seen.add(ip_str)
+        try:
+            ip = ipaddress.ip_address(ip_str)
+        except ValueError:
+            # Defensive: getaddrinfo gave us something we can't parse.
+            # Treat as suspicious rather than as a transient error.
+            raise UrlBlockedPrivateHostError(
+                f"Resolved address '{ip_str}' is not a valid IP."
+            ) from None
+        if _is_non_public_address(ip):
+            raise UrlBlockedPrivateHostError(
+                f"Hostname resolves to non-public address '{ip}'. "
+                f"URL ingestion only accepts public hosts."
+            )
+
+    if not seen:
+        raise UrlFetchFailedError(
+            "Hostname resolved to no usable IPv4 or IPv6 addresses."
+        )
+
+
+def _is_non_public_address(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
+    """True for any address category we refuse to fetch from.
+
+    Includes loopback, link-local, multicast, private RFC1918, reserved,
+    unspecified (0.0.0.0/::). The AWS/GCP cloud metadata endpoint
+    (169.254.169.254) is already caught by ``is_link_local``.
+
+    Also explicitly blocks RFC 6598 Shared Address Space (CGNAT,
+    ``100.64.0.0/10``). In Python's ``ipaddress``, ``is_private``
+    returns False for this range and ``is_reserved`` doesn't cover it,
+    so we check it directly.
+    """
+    if ip.is_loopback or ip.is_link_local or ip.is_multicast:
+        return True
+    if ip.is_private or ip.is_reserved or ip.is_unspecified:
+        return True
+    if isinstance(ip, ipaddress.IPv4Address) and ip in _CGNAT_RFC6598:
+        return True
+    return False
+
+
+# RFC 6598 Shared Address Space — carrier-grade NAT range. Not covered
+# by ``ipaddress.IPv4Address.is_private``.
+_CGNAT_RFC6598: Final[ipaddress.IPv4Network] = ipaddress.ip_network(
+    "100.64.0.0/10"
+)
+
+
+# ----------------------------------------------------------------------
+# Response parsing helpers
+# ----------------------------------------------------------------------
+
+
+def _parse_content_length(header_value: str | None) -> int | None:
+    """Best-effort parse of ``Content-Length``. None if absent/invalid."""
+    if header_value is None:
+        return None
+    try:
+        value = int(header_value.strip())
+    except (ValueError, AttributeError):
+        return None
+    return value if value >= 0 else None
+
+
+def _extract_filename(url_path: str) -> str:
+    """Pull the last path segment as the filename, with a safe fallback."""
+    if not url_path:
+        return "url_audio.bin"
+    # Strip trailing slashes and pick the final segment.
+    segment = unquote(url_path.rstrip("/").split("/")[-1])
+    # Guard against pathological cases (e.g. empty, or hostile path
+    # traversal like '..' — we never use this as a filesystem path, but
+    # also no need to pass it through).
+    if not segment or segment in {".", ".."}:
+        return "url_audio.bin"
+    return segment
+
+
+def _pick_mime_type(content_type_header: str, url_path: str) -> str:
+    """Prefer the response Content-Type; fall back to filename guess."""
+    # Strip charset / parameters: "audio/mpeg; charset=binary" -> "audio/mpeg".
+    primary = (content_type_header or "").split(";")[0].strip().lower()
+    if primary and primary not in {
+        "application/octet-stream",
+        "binary/octet-stream",
+    }:
+        return primary
+
+    # Fall back to filename-based sniffing.
+    import mimetypes
+
+    filename = _extract_filename(url_path)
+    guessed, _enc = mimetypes.guess_type(filename)
+    if guessed:
+        return guessed
+    return "application/octet-stream"