feat(api): Track 3.3 — GET /api/analysis-runs/{run_id}/source-audio

apps/backend/ARCHITECTURE.md

@@ -72,6 +72,7 @@ Custom routes:
 - `GET /api/analysis-runs/{run_id}`
 - `DELETE /api/analysis-runs/{run_id}` — owner can delete their own run; operators with `SONIC_ANALYZER_ADMIN_KEY` set can supply `X-Admin-Key` to delete any run. Admin path is closed when the env var is unset.
 - `GET /api/analysis-runs/{run_id}/artifacts` and `…/artifacts/{artifact_id}`
+- `GET /api/analysis-runs/{run_id}/source-audio` — re-serves the original ingested audio for the run. Owner-only (no admin bypass). Saves a round-trip vs looking up the source artifact id first.
 - `GET /api/analysis-runs/{run_id}/export/csv/{field_path}` — CSV export of a Phase 1 time-series field. See [`docs/adr/0001-phase1-json-schema-v1.md`](../../docs/adr/0001-phase1-json-schema-v1.md) and the registry in [`csv_export.py`](csv_export.py).
 - `POST /api/analysis-runs/{run_id}/spectral-enhancements/{kind}`
 - `POST /api/analysis-runs/{run_id}/pitch-note-translations`

apps/backend/server.py

@@ -2271,6 +2271,101 @@ async def get_run_artifact(
     )
 
 
+@app.get(
+    "/api/analysis-runs/{run_id}/source-audio",
+    response_model=None,
+)
+async def get_run_source_audio(
+    run_id: str,
+    x_asa_user_id: str | None = Header(None),
+    x_asa_user_email: str | None = Header(None),
+) -> FileResponse | JSONResponse:
+    """Re-serve the audio file originally ingested for ``run_id``.
+
+    The same bytes are reachable via
+    ``GET /api/analysis-runs/{run_id}/artifacts/{artifact_id}`` if the
+    caller already knows the artifact id, but this route is stable
+    (the path doesn't change per run) and saves a round-trip — the
+    caller does not need to fetch the snapshot first to learn the
+    artifact id. Useful for Phase 2 reruns where the client already
+    has the run id but no longer has the source bytes locally.
+
+    Owner-only. Unlike ``DELETE``, this route does not honor
+    ``X-Admin-Key``: re-serving another user's audio content has a
+    stronger privacy posture than purging another user's run, and the
+    review (Track 3) scoped admin bypass to delete operations.
+
+    Returns:
+        - 200 with the audio file (FileResponse with the original
+          filename in Content-Disposition and the stored MIME type).
+        - 404 ``RUN_NOT_FOUND`` if the run does not exist or is not
+          owned by the requesting user.
+        - 404 ``SOURCE_AUDIO_NOT_FOUND`` if the run exists but has no
+          source_artifact row (should be unreachable in practice; it
+          would indicate a corrupted runtime state).
+        - 404 ``SOURCE_AUDIO_FILE_MISSING`` if the artifact row exists
+          but the underlying file is gone from disk.
+    """
+    user_context = _resolve_route_user_context(x_asa_user_id, x_asa_user_email)
+    if isinstance(user_context, JSONResponse):
+        return user_context
+    runtime = get_analysis_runtime()
+
+    # Ownership check first; this also confirms the run exists.
+    try:
+        runtime.get_run(run_id, owner_user_id=user_context.user_id)
+    except (KeyError, PermissionError):
+        return _run_not_found_response(run_id)
+
+    try:
+        source_artifact = runtime.get_source_artifact(run_id)
+    except KeyError:
+        # Run was found by get_run() above, but the source_artifact_id
+        # column was null or pointed at a missing row. Defensive — not
+        # expected on the normal create_run path. ``retryable`` is
+        # ``False`` because the run state is corrupted; a retry on the
+        # same id will fail identically.
+        return JSONResponse(
+            status_code=404,
+            content={
+                "error": {
+                    "code": "SOURCE_AUDIO_NOT_FOUND",
+                    "message": (
+                        f"Run '{run_id}' has no source audio artifact "
+                        f"registered."
+                    ),
+                    "retryable": False,
+                }
+            },
+        )
+
+    artifact_local_path = runtime.resolve_artifact_local_path(
+        source_artifact.get("path")
+    )
+    if artifact_local_path is None or not artifact_local_path.is_file():
+        return JSONResponse(
+            status_code=404,
+            content={
+                "error": {
+                    "code": "SOURCE_AUDIO_FILE_MISSING",
+                    "message": (
+                        "Source audio file is no longer available on disk."
+                    ),
+                    # The artifact metadata is intact but the bytes are
+                    # gone; an operator would need to re-ingest. A naive
+                    # retry from the client will not recover.
+                    "retryable": False,
+                }
+            },
+        )
+
+    return FileResponse(
+        path=str(artifact_local_path),
+        media_type=source_artifact.get("mimeType", "application/octet-stream"),
+        filename=source_artifact.get("filename", artifact_local_path.name),
+    )
+
+
 @app.get(
     "/api/analysis-runs/{run_id}/export/csv/{field_path}",
     response_model=None,

apps/backend/tests/test_server.py

@@ -4620,6 +4620,195 @@ def test_url_invalid_envelope_is_not_retryable(self) -> None:
         self.assertFalse(payload["error"]["retryable"])
 
 
+class GetRunSourceAudioRouteTests(unittest.TestCase):
+    """Route tests for GET /api/analysis-runs/{run_id}/source-audio.
+
+    The route re-serves the original ingested audio for a run so callers
+    don't have to re-upload for Phase 2 reruns. These tests cover the
+    HTTP shell: ownership enforcement, missing-source handling, and the
+    happy-path body/headers.
+    """
+
+    AUDIO_BYTES = b"FAKE-AUDIO-PAYLOAD"
+
+    def _decode_json_response(self, response) -> dict:
+        return json.loads(response.body.decode("utf-8"))
+
+    def _make_owned_run(self, runtime, *, owner: str = "owner_user") -> str:
+        created = runtime.create_run(
+            filename="reference.mp3",
+            content=self.AUDIO_BYTES,
+            mime_type="audio/mpeg",
+            owner_user_id=owner,
+            analysis_mode="full",
+            pitch_note_mode="off",
+            pitch_note_backend="auto",
+            interpretation_mode="off",
+            interpretation_profile="producer_summary",
+            interpretation_model=None,
+        )
+        return created["runId"]
+
+    def test_serves_audio_with_filename_and_mime(self) -> None:
+        from analysis_runtime import AnalysisRuntime
+
+        with tempfile.TemporaryDirectory(prefix="asa_source_audio_") as temp_dir:
+            runtime = AnalysisRuntime(Path(temp_dir) / "runtime")
+            run_id = self._make_owned_run(runtime)
+
+            with patch.object(server, "get_analysis_runtime", return_value=runtime), \
+                 patch.dict(
+                     server.os.environ,
+                     {"SONIC_ANALYZER_RUNTIME_PROFILE": "hosted"},
+                     clear=False,
+                 ):
+                response = asyncio.run(
+                    server.get_run_source_audio(
+                        run_id,
+                        x_asa_user_id="owner_user",
+                    )
+                )
+
+            # FileResponse points at the on-disk artifact; verify metadata.
+            self.assertEqual(response.status_code, 200)
+            # FileResponse.media_type is the configured mime
+            self.assertEqual(response.media_type, "audio/mpeg")
+            # The Content-Disposition filename comes from the original
+            # upload filename, not the stored path.
+            disposition = response.headers.get("content-disposition", "")
+            self.assertIn("reference.mp3", disposition)
+
+    def test_non_owner_request_returns_run_not_found(self) -> None:
+        """A different user MUST NOT be able to fetch another user's
+        source audio. Identical 404 envelope to a missing run — the
+        route does not disclose whether the run exists.
+        """
+        from analysis_runtime import AnalysisRuntime
+
+        with tempfile.TemporaryDirectory(prefix="asa_source_audio_") as temp_dir:
+            runtime = AnalysisRuntime(Path(temp_dir) / "runtime")
+            run_id = self._make_owned_run(runtime, owner="owner_user")
+
+            with patch.object(server, "get_analysis_runtime", return_value=runtime), \
+                 patch.dict(
+                     server.os.environ,
+                     {"SONIC_ANALYZER_RUNTIME_PROFILE": "hosted"},
+                     clear=False,
+                 ):
+                response = asyncio.run(
+                    server.get_run_source_audio(
+                        run_id,
+                        x_asa_user_id="not_the_owner",
+                    )
+                )
+
+        self.assertEqual(response.status_code, 404)
+        payload = self._decode_json_response(response)
+        self.assertEqual(payload["error"]["code"], "RUN_NOT_FOUND")
+
+    def test_unknown_run_returns_404(self) -> None:
+        from analysis_runtime import AnalysisRuntime
+
+        with tempfile.TemporaryDirectory(prefix="asa_source_audio_") as temp_dir:
+            runtime = AnalysisRuntime(Path(temp_dir) / "runtime")
+            with patch.object(server, "get_analysis_runtime", return_value=runtime), \
+                 patch.dict(
+                     server.os.environ,
+                     {"SONIC_ANALYZER_RUNTIME_PROFILE": "hosted"},
+                     clear=False,
+                 ):
+                response = asyncio.run(
+                    server.get_run_source_audio(
+                        "does-not-exist",
+                        x_asa_user_id="some_user",
+                    )
+                )
+
+        self.assertEqual(response.status_code, 404)
+        payload = self._decode_json_response(response)
+        self.assertEqual(payload["error"]["code"], "RUN_NOT_FOUND")
+
+    def test_missing_source_artifact_returns_source_audio_not_found(self) -> None:
+        """If the run row exists but get_source_artifact raises (no
+        source_artifact_id linkage), the route surfaces the dedicated
+        SOURCE_AUDIO_NOT_FOUND code rather than the generic RUN_NOT_FOUND.
+
+        Distinct error codes matter because the recovery path differs:
+        RUN_NOT_FOUND → ownership / typo issue, the client should check
+        the id; SOURCE_AUDIO_NOT_FOUND → server-side corruption, the
+        client cannot recover, an operator needs to look.
+        """
+        from analysis_runtime import AnalysisRuntime
+
+        with tempfile.TemporaryDirectory(prefix="asa_source_audio_") as temp_dir:
+            runtime = AnalysisRuntime(Path(temp_dir) / "runtime")
+            run_id = self._make_owned_run(runtime, owner="owner_user")
+
+            # Patch get_source_artifact to simulate missing-linkage path.
+            with patch.object(server, "get_analysis_runtime", return_value=runtime), \
+                 patch.object(
+                     runtime,
+                     "get_source_artifact",
+                     side_effect=KeyError(f"Run {run_id} is missing its source artifact"),
+                 ), \
+                 patch.dict(
+                     server.os.environ,
+                     {"SONIC_ANALYZER_RUNTIME_PROFILE": "hosted"},
+                     clear=False,
+                 ):
+                response = asyncio.run(
+                    server.get_run_source_audio(
+                        run_id,
+                        x_asa_user_id="owner_user",
+                    )
+                )
+
+        self.assertEqual(response.status_code, 404)
+        payload = self._decode_json_response(response)
+        self.assertEqual(payload["error"]["code"], "SOURCE_AUDIO_NOT_FOUND")
+        # Server-side corruption — naive client retry won't recover.
+        self.assertFalse(payload["error"]["retryable"])
+
+    def test_missing_file_on_disk_returns_source_audio_file_missing(self) -> None:
+        """If the artifact row exists but resolve_artifact_local_path
+        returns None or the file isn't on disk, surface
+        SOURCE_AUDIO_FILE_MISSING. Distinct from SOURCE_AUDIO_NOT_FOUND
+        because the recovery is different: the artifact metadata is
+        intact, only the bytes are gone — operator can re-ingest from
+        the original source.
+        """
+        from analysis_runtime import AnalysisRuntime
+
+        with tempfile.TemporaryDirectory(prefix="asa_source_audio_") as temp_dir:
+            runtime = AnalysisRuntime(Path(temp_dir) / "runtime")
+            run_id = self._make_owned_run(runtime, owner="owner_user")
+
+            with patch.object(server, "get_analysis_runtime", return_value=runtime), \
+                 patch.object(
+                     runtime,
+                     "resolve_artifact_local_path",
+                     return_value=None,
+                 ), \
+                 patch.dict(
+                     server.os.environ,
+                     {"SONIC_ANALYZER_RUNTIME_PROFILE": "hosted"},
+                     clear=False,
+                 ):
+                response = asyncio.run(
+                    server.get_run_source_audio(
+                        run_id,
+                        x_asa_user_id="owner_user",
+                    )
+                )
+
+        self.assertEqual(response.status_code, 404)
+        payload = self._decode_json_response(response)
+        self.assertEqual(payload["error"]["code"], "SOURCE_AUDIO_FILE_MISSING")
+        # File bytes missing — operator must re-ingest; client retry
+        # of the same id will not recover.
+        self.assertFalse(payload["error"]["retryable"])
+
+
 class CsvExportRouteTests(unittest.TestCase):
     """Route tests for GET /api/analysis-runs/{run_id}/export/csv/{field_path}.