diff --git a/.claude/settings.local.json b/.claude/settings.local.json index 92d17472..df990b66 100644 --- a/.claude/settings.local.json +++ b/.claude/settings.local.json @@ -10,7 +10,8 @@ "mcp__726edc52-7fd8-4da5-944f-84f446a0ee01__hub_repo_details", "WebSearch", "WebFetch(domain:firebase.google.com)", - "WebFetch(domain:ai.google.dev)" + "WebFetch(domain:ai.google.dev)", + "Bash(git checkout *)" ] } } diff --git a/.gitignore b/.gitignore index eeec02d5..22ec7817 100644 --- a/.gitignore +++ b/.gitignore @@ -62,6 +62,11 @@ apps/backend/tests/fixtures/estimations/ apps/backend/tests/fixtures/beat_tracks/* !apps/backend/tests/fixtures/beat_tracks/README.md +# Optional MT3 model cache. Multi-GB JAX checkpoint downloaded out-of-band +# via `gsutil cp gs://mt3/checkpoints/mt3 apps/backend/models/mt3/` — never +# redistributed and not on the request path. See apps/backend/mt3_transcription.py. +apps/backend/models/ + # Ephemeral local state .claude/scheduled_tasks.lock .continue.sh diff --git a/CLAUDE.md b/CLAUDE.md index 0359ed3b..60d13352 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -137,6 +137,7 @@ The backend supports staged execution via `analysis_runtime.py`, which persists 1. **measurement** — Phase 1 DSP via `analyze.py` 2. **pitch/note translation** — pitch/note extraction on Demucs-separated stems 3. **interpretation** — Gemini Phase 2 advisory +4. **mt3** *(optional, peer of pitch/note translation)* — MT3 polyphonic transcription, gated on `mt3_mode=enabled` per run; emits per-stem MIDI as artifacts. Additive only — never overrides measurement (PURPOSE.md invariant #1). Run by `_execute_mt3_attempt`/`_mt3_worker_loop` in `server.py`. See [`JSON_SCHEMA.md` "Optional MT3 Namespace"](apps/backend/JSON_SCHEMA.md#optional-mt3-namespace). Run-oriented endpoints (`/api/analysis-runs*`) are the canonical interface for staged execution. Legacy `POST /api/analyze`, `POST /api/analyze/estimate`, and `POST /api/phase2` remain only as temporary compatibility wrappers — do not build new functionality on them. diff --git a/apps/backend/JSON_SCHEMA.md b/apps/backend/JSON_SCHEMA.md index e4394e34..49953921 100644 --- a/apps/backend/JSON_SCHEMA.md +++ b/apps/backend/JSON_SCHEMA.md @@ -135,6 +135,11 @@ Current server behavior that affects schema expectations: - `pitchDetail` is only populated when `--separate` is used; requires torchcrepe and separated stems - `danceability` is forwarded as the raw object shown below, not as a scalar - `dsp_json_override` is accepted by the server but does not alter the analyzer payload +- `transcription` (the optional MT3 namespace) is *absent* — not `null` — unless the + env var `ASA_ENABLE_MT3=1` is set AND the MT3 extra is installed AND the call + succeeds. The key is intentionally NOT in `EXPECTED_TOP_LEVEL_KEYS` (the + shared full/fast contract); presence-or-absence is the contract. See the + "Optional MT3 Namespace" section below. --- @@ -806,6 +811,105 @@ Type: `object \| null` --- +## Optional MT3 Namespace + +`transcription.mt3` is an *additive*, opt-in polyphonic-transcription +namespace, produced by Magenta's MT3 (Multi-Task Multitrack Music +Transcription) checkpoint via [`mt3_transcription.py`](./mt3_transcription.py). +It is **purely additive to Phase 1** — it does NOT override or refine +Essentia chord/key/beat/melody outputs (PURPOSE.md invariant #1, "Phase 1 +measurements are ground truth"). Phase 2 may *read* it for richer note +context, but Phase 1's authoritative measurements remain the ground truth. + +### Two emission paths + +MT3 reaches clients through one of two paths: + +1. **Staged runtime** (production path). MT3 runs as its own stage in + [`analysis_runtime.py`](./analysis_runtime.py), peer to + `pitch_note_translation`. Opt in via the `mt3_mode=enabled` form + field on `POST /api/analysis-runs`, or enqueue after-the-fact via + `POST /api/analysis-runs/{run_id}/mt3-transcriptions`. The stage + result lives on `snapshot.stages.mt3.result`. MIDI bytes are + persisted as per-stem artifacts; the result carries `midiArtifactId` + + `midiSizeBytes` per track (see "Schema" below). + +2. **Direct CLI** (legacy / one-off). Running + `./venv/bin/python analyze.py ` with `ASA_ENABLE_MT3=1` set + inlines MT3 results under a top-level `transcription` key in stdout + JSON. In this path each track carries an inline `midiB64` base64 + blob rather than an artifact ref — the CLI has no artifact store. + **Not used by the staged runtime** (which never exports the env var + to the subprocess); `analysis_runtime.complete_measurement()` defensively + pops `transcription` from the measurement-stage result so an operator + who exports the env var globally cannot leak it into the + `stages.measurement` payload. + +Presence contract (load-bearing): + +- For the **staged runtime path**, `snapshot.stages.mt3` is always + present with a status field; it sits at `not_requested` when + `mt3_mode='off'` (the default). +- For the **direct CLI path**, the entire `transcription` key is + **absent from the JSON output** unless `ASA_ENABLE_MT3=1` AND the MT3 + extra is installed AND the call succeeds. +- The key is NOT in `EXPECTED_TOP_LEVEL_KEYS` (the shared full/fast + schema test) by design — adding it there would force the field to + always appear in CLI output, breaking the "absent when off" contract. +- Fast mode (`--fast`) never emits the key (the fast pipeline builds + its own output dict separately, bypassing the MT3 hook entirely). +- The `--pitch-note-only` subprocess mode exits before reaching the + MT3 hook, so the staged pitch/note translation request path is + unaffected. + +### One-time host setup (not run in CI) + +```bash +./apps/backend/venv/bin/pip install -r apps/backend/requirements-mt3.txt +mkdir -p apps/backend/models/mt3 +gsutil -m cp -r gs://mt3/checkpoints/mt3 apps/backend/models/mt3/ + +# Staged runtime usage (per-run opt-in via the HTTP route): +curl -X POST http://127.0.0.1:8100/api/analysis-runs \ + -F track=@ -F mt3_mode=enabled + +# Direct CLI usage (legacy): +export ASA_ENABLE_MT3=1 +./venv/bin/python analyze.py --yes +``` + +### Schema (staged runtime — `snapshot.stages.mt3.result`) + +| Field | Type | Description | +|---|---|---| +| `version` | `string` | Pinned identifier — format `"mt3-py-+"`, e.g. `"mt3-py-0.1.0+magenta-mt3-base"`. Phase 2 reads this verbatim to know what produced the notes. | +| `stemsUsed` | `string[]` | Stems MT3 actually transcribed. Values are Demucs canonical names (`"bass"`, `"other"`, `"vocals"`) or `["full_mix"]` when no stems were provided. | +| `tracks` | `Mt3Track[]` | One track per stem MT3 successfully processed. May be shorter than `stemsUsed` if a per-stem failure was caught and logged. | +| `tracks[*].instrument` | `string` | Stem name (matches one entry in `stemsUsed`). | +| `tracks[*].midiArtifactId` | `string \| null` | Artifact ID for the per-stem MIDI bytes. Fetch via `GET /api/analysis-runs/{run_id}/artifacts/{midiArtifactId}` (mime `audio/midi`). Null only when the track had no MIDI body (defensive — should not happen in practice). | +| `tracks[*].midiSizeBytes` | `int` | Size of the MIDI artifact in bytes. | +| `tracks[*].noteCount` | `int` | Number of notes in the decoded MIDI. | +| `tracks[*].pitchRange` | `[int, int]` | `[minMidi, maxMidi]` across all notes in this track. `[0, 0]` when empty. | + +### Direct CLI shape difference + +When the legacy `ASA_ENABLE_MT3=1` env-var path runs, each track carries +an inline `midiB64: string` (base64-encoded MIDI bytes) instead of the +artifact-ref pair. The staged-runtime executor in +[`server.py::_execute_mt3_attempt`](./server.py) reads that `midiB64`, +writes the bytes as a `mt3_track_` artifact, and rewrites +the track with `midiArtifactId` + `midiSizeBytes` before persisting. +This is the intentional cross-layer mapping — Python and TS describe +the same per-stem concept at different layers. + +LLM interpretation note: MT3 is a multi-instrument AMT system trained +on polyphonic material. Treat MIDI output as a *transcription hint* +layered on top of Phase 1, not a substitute for the deterministic +measurements. Do not "vote" between MT3 notes and Phase 1's chord/key +estimates — Phase 1 wins by invariant. + +--- + ## Additional Notes for LLM Consumers 1. Treat low-confidence outputs as hints, not truth: diff --git a/apps/backend/analysis_runtime.py b/apps/backend/analysis_runtime.py index 44b7c663..909b9138 100644 --- a/apps/backend/analysis_runtime.py +++ b/apps/backend/analysis_runtime.py @@ -10,10 +10,23 @@ from artifact_storage import ArtifactStorage, FilesystemArtifactStorage import uuid +# Single source of truth for the MT3 checkpoint identifier — used as the +# default `checkpoint_id` column value when queuing an mt3 attempt. The +# executor overrides this with whatever transcribe() actually ran with so +# the recorded value reflects the run, but having a default here lets the +# runtime emit queued snapshots before the worker has picked the row up. +# Import is module-level-safe: mt3_transcription.py top-level is stdlib +# + dataclass only; JAX/t5x are lazy inside transcribe(). +from mt3_transcription import MT3_CHECKPOINT_ID as DEFAULT_MT3_CHECKPOINT_ID + SQLITE_BUSY_TIMEOUT_MS = 5_000 MEASUREMENT_PIPELINE_PROGRESS_STATUSES = {"pending", "running", "completed"} LOCAL_RUNTIME_OWNER_USER_ID = "local-dev" +# Stage-status values for the MT3 stage. Mirrors pitch_note_translation: +# {queued, running, completed, failed, interrupted}. Added to the public +# snapshot via _mt3_stage_snapshot below. + def _utc_now_iso() -> str: return datetime.now(UTC).isoformat() @@ -43,6 +56,18 @@ def __init__(self, pitch_note_backend: str): super().__init__(f"Unsupported pitch/note backend '{pitch_note_backend}'.") +class UnsupportedMt3ModeError(ValueError): + """Raised when the create-run request specifies an unknown mt3_mode. + + The route layer (server.py) catches this and surfaces it as a typed + 400 ``MT3_MODE_UNSUPPORTED`` response. Mirror of UnsupportedPitchNoteModeError. + """ + + def __init__(self, mt3_mode: str): + self.mt3_mode = mt3_mode + super().__init__(f"Unsupported mt3 mode '{mt3_mode}'.") + + class AnalysisRuntime: def __init__( self, @@ -140,6 +165,19 @@ def _ensure_schema(self) -> None: created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); + + CREATE TABLE IF NOT EXISTS mt3_attempts ( + id TEXT PRIMARY KEY, + run_id TEXT NOT NULL, + checkpoint_id TEXT NOT NULL, + status TEXT NOT NULL, + result_json TEXT, + provenance_json TEXT, + diagnostics_json TEXT, + error_json TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); """ ) self._ensure_column( @@ -181,6 +219,28 @@ def _ensure_schema(self) -> None: "grounded_pitch_note_attempt_id", "TEXT", ) + # MT3 stage columns. Default to 'off' so existing local DBs that + # predate the staged MT3 stage continue to opt out cleanly — + # matches the `requested_analysis_mode = 'full'` backfill above. + self._ensure_column( + conn, + "analysis_runs", + "requested_mt3_mode", + "TEXT", + ) + self._ensure_column( + conn, + "analysis_runs", + "preferred_mt3_attempt_id", + "TEXT", + ) + conn.execute( + """ + UPDATE analysis_runs + SET requested_mt3_mode = 'off' + WHERE requested_mt3_mode IS NULL + """ + ) @staticmethod def _ensure_column( @@ -211,8 +271,13 @@ def create_run( interpretation_model: str | None, legacy_request_id: str | None = None, analysis_mode: str = "full", + mt3_mode: str = "off", expose_source_path_in_snapshot: bool = False, ) -> dict[str, Any]: + # Validate mt3_mode here so a typed error reaches the route layer + # rather than silently inserting an unknown enum value. + if mt3_mode not in {"off", "enabled"}: + raise UnsupportedMt3ModeError(mt3_mode) artifact_id = str(uuid4()) created_at = _utc_now_iso() stored_artifact = self.artifact_storage.store_bytes( @@ -242,10 +307,11 @@ def create_run( requested_interpretation_mode, requested_interpretation_profile, requested_interpretation_model, + requested_mt3_mode, legacy_request_id, created_at, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( run_id, @@ -257,6 +323,7 @@ def create_run( interpretation_mode, interpretation_profile, interpretation_model, + mt3_mode, legacy_request_id, created_at, created_at, @@ -494,9 +561,18 @@ def get_interpretation_grounding(self, run_id: str) -> dict[str, Any]: """, (run_id,), ).fetchall() + mt3_rows = conn.execute( + """ + SELECT * FROM mt3_attempts + WHERE run_id = ? + ORDER BY created_at DESC + """, + (run_id,), + ).fetchall() if run_row is None or measurement_row is None: raise KeyError(f"Unknown run {run_id}") preferred_pitch_note = self._preferred_pitch_note_row(run_row, pitch_note_rows) + preferred_mt3 = self._preferred_mt3_row(run_row, mt3_rows) return { "measurementOutputId": str(measurement_row["id"]), "measurementStatus": str(measurement_row["status"]), @@ -512,6 +588,22 @@ def get_interpretation_grounding(self, run_id: str) -> dict[str, Any]: if preferred_pitch_note is not None else None ), + # MT3 grounding. Optional like pitchNote — present only when + # an MT3 attempt has completed for the run. Phase 2 receives + # this verbatim in OPTIONAL_MT3_TRANSCRIPTION_RESULT_JSON + # (see _build_phase2_prompt). Additive only: PURPOSE.md + # invariant #1 — Phase 1 is ground truth. + "mt3AttemptId": ( + str(preferred_mt3["id"]) if preferred_mt3 is not None else None + ), + "mt3Status": ( + str(preferred_mt3["status"]) if preferred_mt3 is not None else "not_requested" + ), + "mt3Result": ( + _json_loads(preferred_mt3["result_json"]) + if preferred_mt3 is not None + else None + ), } def get_run(self, run_id: str, *, owner_user_id: str | None = None) -> dict[str, Any]: @@ -553,12 +645,30 @@ def get_run(self, run_id: str, *, owner_user_id: str | None = None) -> dict[str, """, (run_id,), ).fetchall() + mt3_rows = conn.execute( + """ + SELECT * FROM mt3_attempts + WHERE run_id = ? + ORDER BY created_at DESC + """, + (run_id,), + ).fetchall() preferred_pitch_note = self._preferred_pitch_note_row(run_row, pitch_note_rows) preferred_interpretation = self._preferred_interpretation_row( run_row, interpretation_rows ) + preferred_mt3 = self._preferred_mt3_row(run_row, mt3_rows) measurement_status = measurement_row["status"] + # requested_mt3_mode may be NULL on rows that predate the column + # backfill (the _ensure_column UPDATE handles most cases but a race + # at app start can leave one through). Default to "off" so the + # snapshot is always coherent. + requested_mt3_mode = ( + run_row["requested_mt3_mode"] + if run_row["requested_mt3_mode"] is not None + else "off" + ) return { "runId": run_id, @@ -569,6 +679,7 @@ def get_run(self, run_id: str, *, owner_user_id: str | None = None) -> dict[str, "interpretationMode": run_row["requested_interpretation_mode"], "interpretationProfile": run_row["requested_interpretation_profile"], "interpretationModel": run_row["requested_interpretation_model"], + "mt3Mode": requested_mt3_mode, }, "artifacts": { "sourceAudio": { @@ -608,6 +719,12 @@ def get_run(self, run_id: str, *, owner_user_id: str | None = None) -> dict[str, preferred_interpretation, interpretation_rows, ), + "mt3": self._mt3_stage_snapshot( + requested_mt3_mode, + measurement_status, + preferred_mt3, + mt3_rows, + ), }, } @@ -683,6 +800,13 @@ def complete_measurement( # Strip transcriptionDetail — it's a Layer 2 (pitch/note translation) concern. # The pitch/note translation worker produces this independently through its own stage. measurement_result.pop("transcriptionDetail", None) + # Strip transcription — same story for the MT3 polyphonic-transcription + # namespace. analyze.py's legacy env-var hook (ASA_ENABLE_MT3=1) may + # still emit it for direct CLI users, but the staged runtime owns + # this stage now and stores it under stages.mt3 instead. The pop is + # defense-in-depth so an operator who exports the env var globally + # doesn't leak a stale `transcription` key into the measurement row. + measurement_result.pop("transcription", None) self._update_measurement_row( run_id, status="completed", @@ -868,6 +992,167 @@ def fail_pitch_note_attempt( error=error, ) + # ─── MT3 polyphonic-transcription stage ──────────────────────────── + # Peer of pitch_note_translation. Gated only on measurement completion + # (NOT on pitch_note completion) so the two can run concurrently. See + # docs/ARCHITECTURE_STRATEGY.md and mt3_transcription.py for the + # additive-only invariant: MT3 output never feeds back into the + # measurement result. + + def create_mt3_attempt( + self, + run_id: str, + *, + checkpoint_id: str = DEFAULT_MT3_CHECKPOINT_ID, + status: str = "queued", + result: dict[str, Any] | None = None, + provenance: dict[str, Any] | None = None, + ) -> str: + attempt_id = str(uuid4()) + now = _utc_now_iso() + with self._connect() as conn: + conn.execute( + """ + INSERT INTO mt3_attempts ( + id, + run_id, + checkpoint_id, + status, + result_json, + provenance_json, + diagnostics_json, + error_json, + created_at, + updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + attempt_id, + run_id, + checkpoint_id, + status, + _json_dumps(result), + _json_dumps(provenance), + None, + None, + now, + now, + ), + ) + if status == "completed": + conn.execute( + """ + UPDATE analysis_runs + SET preferred_mt3_attempt_id = ?, updated_at = ? + WHERE id = ? + """, + (attempt_id, now, run_id), + ) + return attempt_id + + def reserve_mt3_attempt(self, attempt_id: str) -> bool: + now = _utc_now_iso() + with self._connect() as conn: + cursor = conn.execute( + """ + UPDATE mt3_attempts + SET status = 'running', updated_at = ? + WHERE id = ? AND status = 'queued' + """, + (now, attempt_id), + ) + return cursor.rowcount > 0 + + def reserve_next_mt3_attempt(self) -> dict[str, Any] | None: + now = _utc_now_iso() + with self._connect() as conn: + row = conn.execute( + """ + SELECT ma.id, ma.run_id, ma.checkpoint_id + FROM mt3_attempts ma + JOIN measurement_outputs mo ON mo.run_id = ma.run_id + WHERE ma.status = 'queued' AND mo.status = 'completed' + ORDER BY ma.created_at ASC + LIMIT 1 + """ + ).fetchone() + if row is None: + return None + cursor = conn.execute( + """ + UPDATE mt3_attempts + SET status = 'running', updated_at = ? + WHERE id = ? AND status = 'queued' + """, + (now, row["id"]), + ) + if cursor.rowcount == 0: + return None + return { + "attemptId": row["id"], + "runId": row["run_id"], + "checkpointId": row["checkpoint_id"], + } + + def complete_mt3_attempt( + self, + attempt_id: str, + *, + result: dict[str, Any] | None, + provenance: dict[str, Any] | None, + diagnostics: dict[str, Any] | None = None, + ) -> None: + now = _utc_now_iso() + with self._connect() as conn: + attempt_row = conn.execute( + "SELECT run_id FROM mt3_attempts WHERE id = ?", + (attempt_id,), + ).fetchone() + if attempt_row is None: + raise KeyError(f"Unknown mt3 attempt {attempt_id}") + conn.execute( + """ + UPDATE mt3_attempts + SET status = ?, result_json = ?, provenance_json = ?, diagnostics_json = ?, error_json = ?, updated_at = ? + WHERE id = ? + """, + ( + "completed", + _json_dumps(result), + _json_dumps(provenance), + _json_dumps(diagnostics), + None, + now, + attempt_id, + ), + ) + conn.execute( + """ + UPDATE analysis_runs + SET preferred_mt3_attempt_id = ?, updated_at = ? + WHERE id = ? + """, + (attempt_id, now, attempt_row["run_id"]), + ) + + def fail_mt3_attempt( + self, + attempt_id: str, + *, + error: dict[str, Any], + provenance: dict[str, Any] | None = None, + diagnostics: dict[str, Any] | None = None, + ) -> None: + self._update_attempt_row( + table="mt3_attempts", + attempt_id=attempt_id, + status="failed", + result=None, + provenance=provenance, + diagnostics=diagnostics, + error=error, + ) + def create_interpretation_attempt( self, run_id: str, @@ -938,12 +1223,34 @@ def reserve_interpretation_attempt(self, attempt_id: str) -> bool: def reserve_next_interpretation_attempt(self) -> dict[str, Any] | None: now = _utc_now_iso() with self._connect() as conn: + # Interpretation is the LAST stage: it consumes the additive + # grounding produced by the mt3 and pitch_note peers (Gemini reads + # mt3Result / pitchNoteResult from get_interpretation_grounding in + # _execute_interpretation_attempt). Those peers race interpretation + # and the slower one (MT3 loads multi-GB JAX weights) usually loses, + # so without this guard the grounding is almost always None by the + # time interpretation runs. Block a queued interpretation while + # either grounding stage is still in-flight ('queued'/'running'); + # the guard reads no requested_*_mode because no in-flight rows means + # the stage either wasn't requested or already settled. Terminal + # states (completed/failed/interrupted) are not in-flight, so a + # failure/interrupt/restart-recovery unblocks interpretation, which + # then runs with whatever grounding is available (possibly None). row = conn.execute( """ SELECT ia.id, ia.run_id, ia.profile_id, ia.model_name FROM interpretation_attempts ia JOIN measurement_outputs mo ON mo.run_id = ia.run_id - WHERE ia.status = 'queued' AND mo.status = 'completed' + WHERE ia.status = 'queued' + AND mo.status = 'completed' + AND NOT EXISTS ( + SELECT 1 FROM mt3_attempts ma + WHERE ma.run_id = ia.run_id AND ma.status IN ('queued', 'running') + ) + AND NOT EXISTS ( + SELECT 1 FROM pitch_note_translation_attempts pna + WHERE pna.run_id = ia.run_id AND pna.status IN ('queued', 'running') + ) ORDER BY ia.created_at ASC LIMIT 1 """ @@ -1150,6 +1457,14 @@ def recover_incomplete_attempts(self) -> None: """, (now,), ) + conn.execute( + """ + UPDATE mt3_attempts + SET status = 'interrupted', updated_at = ? + WHERE status = 'running' + """, + (now,), + ) def interrupt_run(self, run_id: str) -> dict[str, Any]: now = _utc_now_iso() @@ -1208,11 +1523,25 @@ def interrupt_run(self, run_id: str) -> dict[str, Any]: """, (interruption_diagnostics, now, run_id), ) + conn.execute( + """ + UPDATE mt3_attempts + SET status = 'interrupted', + result_json = NULL, + provenance_json = NULL, + diagnostics_json = ?, + error_json = NULL, + updated_at = ? + WHERE run_id = ? AND status IN ('queued', 'running', 'completed', 'failed') + """, + (interruption_diagnostics, now, run_id), + ) conn.execute( """ UPDATE analysis_runs SET preferred_pitch_note_attempt_id = NULL, preferred_interpretation_attempt_id = NULL, + preferred_mt3_attempt_id = NULL, updated_at = ? WHERE id = ? """, @@ -1241,6 +1570,7 @@ def delete_run(self, run_id: str) -> None: ).fetchall() conn.execute("DELETE FROM interpretation_attempts WHERE run_id = ?", (run_id,)) conn.execute("DELETE FROM pitch_note_translation_attempts WHERE run_id = ?", (run_id,)) + conn.execute("DELETE FROM mt3_attempts WHERE run_id = ?", (run_id,)) conn.execute("DELETE FROM measurement_outputs WHERE run_id = ?", (run_id,)) conn.execute("DELETE FROM run_artifacts WHERE run_id = ?", (run_id,)) conn.execute("DELETE FROM analysis_runs WHERE id = ?", (run_id,)) @@ -1492,6 +1822,10 @@ def _enqueue_requested_followups(self, run_id: str) -> None: "SELECT 1 FROM interpretation_attempts WHERE run_id = ? LIMIT 1", (run_id,), ).fetchone() + mt3_exists = conn.execute( + "SELECT 1 FROM mt3_attempts WHERE run_id = ? LIMIT 1", + (run_id,), + ).fetchone() if ( run_row["requested_pitch_note_mode"] != "off" @@ -1511,6 +1845,32 @@ def _enqueue_requested_followups(self, run_id: str) -> None: }, ) + # MT3 enqueue. requested_mt3_mode is read via .get-like fallback + # because old SQLite rows may have NULL there (the _ensure_column + # backfill above writes 'off', but defense-in-depth lets us survive + # a race where a brand-new run is created right at the seam). + requested_mt3_mode = ( + run_row["requested_mt3_mode"] + if run_row["requested_mt3_mode"] is not None + else "off" + ) + if requested_mt3_mode != "off" and mt3_exists is None: + self.create_mt3_attempt( + run_id, + checkpoint_id=DEFAULT_MT3_CHECKPOINT_ID, + status="queued", + provenance={"checkpointId": DEFAULT_MT3_CHECKPOINT_ID}, + ) + + # Interpretation MUST be enqueued LAST — after the pitch_note and mt3 + # blocks above. reserve_next_interpretation_attempt blocks a queued + # interpretation while an mt3/pitch_note attempt is in-flight + # ('queued'/'running'), but each create_*_attempt commits in its own + # transaction. Were the interpretation row created before the grounding + # rows, the 0.25s-poll interpretation worker could reserve an ungrounded + # attempt in the window before the grounding row commits — silently + # defeating the gate for the mt3-on/pitch_note-off case. Creating it last + # guarantees the grounding rows already exist whenever interpretation does. if ( run_row["requested_interpretation_mode"] != "off" and interpretation_exists is None @@ -1581,6 +1941,25 @@ def _preferred_pitch_note_row( return row return pitch_note_rows[0] if pitch_note_rows else None + @staticmethod + def _preferred_mt3_row( + run_row: sqlite3.Row, mt3_rows: list[sqlite3.Row] + ) -> sqlite3.Row | None: + # Mirror _preferred_pitch_note_row: explicit preference wins, else + # newest attempt. _ensure_column may not have populated + # preferred_mt3_attempt_id on legacy rows; the column access is + # defensive (sqlite3.Row raises IndexError on missing columns, + # so we read via try/except rather than [].). + try: + preferred_id = run_row["preferred_mt3_attempt_id"] + except (IndexError, KeyError): + preferred_id = None + if preferred_id: + for row in mt3_rows: + if row["id"] == preferred_id: + return row + return mt3_rows[0] if mt3_rows else None + @staticmethod def _preferred_interpretation_row( run_row: sqlite3.Row, interpretation_rows: list[sqlite3.Row] @@ -1592,6 +1971,45 @@ def _preferred_interpretation_row( return row return interpretation_rows[0] if interpretation_rows else None + @staticmethod + def _mt3_stage_snapshot( + requested_mode: str, + measurement_status: str, + preferred_row: sqlite3.Row | None, + rows: list[sqlite3.Row], + ) -> dict[str, Any]: + """Mirror of _pitch_note_stage_snapshot. Always non-authoritative — + MT3 output is purely additive and never overrides Phase 1 (PURPOSE.md + invariant #1, "Phase 1 measurements are ground truth").""" + if preferred_row is not None: + status = preferred_row["status"] + elif requested_mode == "off": + status = "not_requested" + elif measurement_status == "interrupted": + status = "interrupted" + elif measurement_status == "completed": + status = "ready" + else: + status = "blocked" + + return { + "status": status, + "authoritative": False, + "preferredAttemptId": preferred_row["id"] if preferred_row is not None else None, + "attemptsSummary": [ + { + "attemptId": row["id"], + "checkpointId": row["checkpoint_id"], + "status": row["status"], + } + for row in rows + ], + "result": _json_loads(preferred_row["result_json"]) if preferred_row is not None else None, + "provenance": _json_loads(preferred_row["provenance_json"]) if preferred_row is not None else None, + "diagnostics": _json_loads(preferred_row["diagnostics_json"]) if preferred_row is not None else None, + "error": _json_loads(preferred_row["error_json"]) if preferred_row is not None else None, + } + @staticmethod def _pitch_note_stage_snapshot( requested_mode: str, diff --git a/apps/backend/analyze.py b/apps/backend/analyze.py index 0c939f0f..6463cd1f 100644 --- a/apps/backend/analyze.py +++ b/apps/backend/analyze.py @@ -1178,6 +1178,65 @@ def _run_pitch_note_translation( shutil.rmtree(temp_dir, ignore_errors=True) +def _run_mt3_transcription( + audio_path: str, + stem_dir: str | None = None, + stem_output_dir: str | None = None, +) -> None: + """Run MT3 polyphonic transcription only and print JSON to stdout. + + Companion to :func:`_run_pitch_note_translation`. Used by the analyse + server's MT3 stage executor (`_execute_mt3_attempt` in server.py) so + JAX/t5x memory is freed when the subprocess exits. + + Stems handover: + - If ``stem_dir`` is provided and contains canonical Demucs + stems (bass.wav / other.wav / vocals.wav), MT3 consumes those + directly — no Demucs invocation needed. This is the path taken + when pitch_note ran first and persisted its stems. + - If ``stem_dir`` is missing AND ``stem_output_dir`` is provided, + run Demucs into ``stem_output_dir`` so the caller can persist + the resulting stems back as artifacts. This is the path taken + when MT3 runs first and pitch_note may follow. + - If both are missing, MT3 falls back to running on the full mix + (via ``_resolve_sources`` inside ``mt3_transcription.transcribe``). + + Errors propagate as non-zero exit + stderr text; the executor matches on + "MT3 backend not installed" / "MT3 checkpoint missing" to distinguish + MT3_NOT_AVAILABLE (retryable=false) from MT3_TRANSCRIPTION_FAILED. + """ + # Lazy imports — keep JAX / t5x out of analyze.py's import graph for + # every code path other than --mt3-only. The pathlib import is also + # deferred for symmetry; nothing else in analyze.py needs Path today. + from pathlib import Path + from mt3_transcription import transcribe as mt3_transcribe + + stems_dir_path: Path | None = None + if stem_dir is not None and os.path.isdir(stem_dir): + stems_dir_path = Path(stem_dir) + + # If no pre-separated stems were provided but the caller asked us to + # write them somewhere, run Demucs into that directory. The caller + # (server.py::_execute_mt3_attempt) reads the resulting files and + # records them as stem_ artifacts so the next stage (pitch_note + # if it runs later) can reuse them. Mirrors --pitch-note-only's + # handover convention so both stages can short-circuit Demucs. + if stems_dir_path is None and stem_output_dir is not None: + os.makedirs(stem_output_dir, exist_ok=True) + separated = separate_stems(audio_path, output_dir=stem_output_dir) + if isinstance(separated, dict) and separated: + # separate_stems returns {"bass": "/path/...wav", ...}. Recover + # the shared parent — all stem files live in one directory. + for stem_path in separated.values(): + if isinstance(stem_path, str) and os.path.isfile(stem_path): + stems_dir_path = Path(stem_path).parent + break + + result = mt3_transcribe(audio_path, stems_dir=stems_dir_path) + json.dump(result.to_payload(), sys.stdout, indent=2) + sys.stdout.write("\n") + + def _run_per_stem_analyses( stems: dict | None, sample_rate: int, @@ -1294,7 +1353,7 @@ def _run_per_stem_analyses( def main(): if len(sys.argv) < 2: print( - "Usage: ./venv/bin/python analyze.py [--separate] [--fast] [--standard] [--transcribe] [--yes] [--pitch-note-only] [--stem-dir DIR] [--stem-output-dir DIR] [--pitch-note-backend BACKEND]", + "Usage: ./venv/bin/python analyze.py [--separate] [--fast] [--standard] [--transcribe] [--yes] [--pitch-note-only] [--mt3-only] [--stem-dir DIR] [--stem-output-dir DIR] [--pitch-note-backend BACKEND]", file=sys.stderr, ) sys.exit(1) @@ -1308,6 +1367,7 @@ def main(): run_transcribe = "--transcribe" in optional_args auto_yes = "--yes" in optional_args pitch_note_only = "--pitch-note-only" in optional_args + mt3_only = "--mt3-only" in optional_args # --pitch-note-only: run pitch/note translation, print JSON, exit if pitch_note_only: @@ -1333,6 +1393,31 @@ def main(): backend_id=backend_id, ) sys.exit(0) + + # --mt3-only: run MT3 polyphonic transcription, print JSON, exit. + # Companion to --pitch-note-only — used by the staged-runtime MT3 + # stage executor (server.py::_execute_mt3_attempt). Stems handover + # is bidirectional: --stem-dir consumes stems pitch_note already + # wrote; --stem-output-dir writes new stems for pitch_note to pick + # up later. Falls back to full-mix MT3 when neither is supplied. + if mt3_only: + stem_dir = None + stem_output_dir = None + if "--stem-dir" in optional_args: + idx = optional_args.index("--stem-dir") + if idx + 1 < len(optional_args): + stem_dir = optional_args[idx + 1] + if "--stem-output-dir" in optional_args: + idx = optional_args.index("--stem-output-dir") + if idx + 1 < len(optional_args): + stem_output_dir = optional_args[idx + 1] + _run_mt3_transcription( + audio_path, + stem_dir=stem_dir, + stem_output_dir=stem_output_dir, + ) + sys.exit(0) + stems = None analysis_estimate = get_audio_duration_seconds(audio_path) @@ -1756,6 +1841,35 @@ def main(): print(f"[warn] stem-first overlay failed: {exc}", file=sys.stderr) result["stemAnalysis"] = None + # Optional MT3 polyphonic transcription pass. Gated on the ASA_ENABLE_MT3 + # env var (default off) so the base ASA install + the standard request + # path never touch MT3 / t5x / JAX. The result lives in its own top-level + # ``transcription`` namespace and is *purely additive* to Phase 1 — it + # does not override or refine any Essentia chord/key/beat/melody output + # (PURPOSE.md invariant #1, "Phase 1 measurements are ground truth"). + # Failures are caught, logged as [warn], and never block the Phase 1 JSON. + if os.getenv("ASA_ENABLE_MT3", "").strip() == "1": + # Lazy import — keeps the mt3_transcription module (and its lazy + # JAX import inside transcribe()) out of the import graph entirely + # when the flag is off. analyze.py itself doesn't need pathlib; + # transcribe() coerces audio_path internally so we can pass the + # raw string straight through. + try: + from mt3_transcription import ( + discover_stems_dir as _mt3_discover_stems_dir, + transcribe as mt3_transcribe, + ) + mt3_stems_dir = _mt3_discover_stems_dir(stems) + print("@@MT3_TRANSCRIPTION_START", file=sys.stderr) + mt3_result = mt3_transcribe(audio_path, stems_dir=mt3_stems_dir) + result["transcription"] = {"mt3": mt3_result.to_payload()} + print("@@MT3_TRANSCRIPTION_COMPLETE", file=sys.stderr) + except Exception as exc: # noqa: BLE001 - MT3 must never block Phase 1 + print(f"[warn] MT3 transcription failed: {exc}", file=sys.stderr) + # Intentionally do NOT set result["transcription"] = None — the + # JSON contract is "absent when the flag is off OR when MT3 fails." + # Adding a null key would make the field always-present. + # Build final output in the exact requested key order output = { "bpm": result.get("bpm"), @@ -1825,6 +1939,14 @@ def main(): "essentiaFeatures": result.get("essentiaFeatures"), } + # Conditional MT3 namespace — *absent* (not null) by default. Only added + # when ASA_ENABLE_MT3=1 AND the transcribe() call succeeded. See the gate + # above. Placed after the literal output dict so the key never appears + # with a null value, which would silently change the contract for every + # caller that introspects `set(payload.keys())`. + if "transcription" in result: + output["transcription"] = result["transcription"] + _emit_progress_marker("complete", "Analysis complete.", 1.0) print("Done.", file=sys.stderr) print(json.dumps(output, indent=2)) diff --git a/apps/backend/analyze_estimate.py b/apps/backend/analyze_estimate.py index 4730dea7..7875d361 100644 --- a/apps/backend/analyze_estimate.py +++ b/apps/backend/analyze_estimate.py @@ -63,6 +63,7 @@ def build_analysis_estimate( run_transcribe: bool, run_fast: bool = False, run_standard: bool = False, + run_mt3: bool = False, ) -> dict: stages = [] @@ -120,6 +121,24 @@ def build_analysis_estimate( } ) + if run_mt3: + # MT3 cost model: high fixed overhead from JAX/t5x model load + # (~30-60s on first call, multi-GB weights), then per-second + # inference ranging ~0.2-0.8x duration on CPU. On a 4-min track + # this lands around 60s-200s; on a 10s clip it's overhead-bound + # at ~60-180s. Numbers tuned to be conservative — operator-facing + # UIs should err high so users aren't surprised mid-run. + mt3_seconds = _estimate_stage_seconds( + duration_seconds, 0.20, 0.80, 60.0, 180.0 + ) + stages.append( + { + "key": "mt3_transcription", + "label": "MT3 polyphonic transcription", + "seconds": mt3_seconds, + } + ) + total_min = sum(stage["seconds"]["min"] for stage in stages) total_max = sum(stage["seconds"]["max"] for stage in stages) diff --git a/apps/backend/mt3_transcription.py b/apps/backend/mt3_transcription.py new file mode 100644 index 00000000..525d96ea --- /dev/null +++ b/apps/backend/mt3_transcription.py @@ -0,0 +1,329 @@ +"""MT3 polyphonic transcription — additive, flag-gated Phase 1 extension. + +This module wraps Magenta's MT3 (Multi-Task Multitrack Music Transcription) +checkpoint as an *optional* secondary transcription stage. It runs only when +the env var ``ASA_ENABLE_MT3`` is set to ``"1"`` (default off) and is purely +additive to Phase 1 — it does NOT override or refine Phase 1's +chord/key/beat/melody measurements (see PURPOSE.md invariant #1, "Phase 1 +measurements are ground truth"). All MT3 output is namespaced under the new +top-level ``transcription`` key in the analyze.py JSON envelope; that key is +*absent* (not null) when the flag is off. + +One-time setup (host machine, not CI): + + # Install the MT3 / t5x / JAX extra into a SEPARATE venv — never the + # product venv (its older numpy/protobuf transitive deps would break + # Essentia/torch on the request path). ASA uses requirements-*.txt files + # in place of pyproject.toml extras (mirroring requirements-eval.txt). See + # CLAUDE.md and the install note atop requirements-mt3.txt. + python3.11 -m venv apps/backend/venv-mt3 + apps/backend/venv-mt3/bin/pip install -r apps/backend/requirements.txt \ + -r apps/backend/requirements-mt3.txt + + # Download the MT3 checkpoint to the model cache (gitignored). + mkdir -p apps/backend/models/mt3 + gsutil -m cp -r gs://mt3/checkpoints/mt3 apps/backend/models/mt3/ + + # Enable for one analyze.py invocation. + export ASA_ENABLE_MT3=1 + +Lazy-import contract +==================== +All MT3 / t5x / JAX deps are imported *inside* ``transcribe()``. Module-level +imports stay limited to the stdlib + ``numpy`` (already a base ASA dep). This +keeps app startup free of multi-second JAX initialization when the flag is +off, and ensures the base ASA install runs unchanged. No other file in the +backend imports ``jax`` — that isolation is load-bearing for keeping JAX/PyTorch +ABI conflicts contained. + +The PyTorch port question +========================= +The goal text says "PyTorch first if a maintained port exists". As of +2026-05, no production-stable PyTorch port of MT3 is published; community +ports are research-quality and unmaintained. This module therefore wraps the +canonical t5x/JAX inference path via ``mt3.inference_model.InferenceModel``. +If a maintained PyTorch port lands later, the swap is contained here. + +Contract emitted to JSON +======================== +``transcribe()`` returns an ``Mt3Result`` dataclass. The Python attributes +are snake_case (per Python convention); the JSON projection through +``Mt3Result.to_payload()`` is camelCase (per CLAUDE.md tripwire #3 — ASA's +JSON envelope is camelCase end-to-end with no conversion layer between +analyze.py and the frontend). The mapping is: + + Python attr → JSON key + ───────────────── ──────────── + version → "version" + stems_used → "stemsUsed" + tracks[*].midi_b64 → "midiB64" + tracks[*].note_count → "noteCount" + tracks[*].pitch_range → "pitchRange" + +Phase 2 reads the ``version`` string verbatim to know what produced the +notes — keep ``MT3_CHECKPOINT_ID`` in sync with the checkpoint actually +loaded. +""" + +from __future__ import annotations + +import base64 +import os +import sys +from dataclasses import dataclass, field +from io import BytesIO +from pathlib import Path +from typing import Iterable + +# Pinned checkpoint identifier — surfaces in Mt3Result.version so Phase 2 +# (and any downstream symbolic consumer) knows exactly what produced the notes. +# This is a best-effort identifier: Magenta does not publish a content-hash +# or release-tag for the checkpoint at gs://mt3/checkpoints/mt3 that we can +# verify automatically. Operators downloading newer revisions should append +# a date or hash they record at download time, e.g. +# ``"magenta-mt3-base@2026-05-28"``. The format is checked by tests, so +# the suffix is optional but its shape is constrained when present. +MT3_CHECKPOINT_ID = "magenta-mt3-base" +MT3_MODULE_VERSION = "0.1.0" + +# Default cache location for MT3 weights — gitignored via apps/backend/models/. +# Resolved relative to this module so it survives different cwd conventions. +DEFAULT_MT3_CHECKPOINT_DIR = ( + Path(__file__).resolve().parent / "models" / "mt3" +) + +# Stem filenames Demucs writes (see analyze_audio_io.separate_stems). MT3 runs +# best on instrumental stems; we skip drums by default because MT3's drum head +# is weaker than purpose-built drum trackers and Phase 1 already covers drums +# in detail (snareDetail, hihatDetail, kickDetail). +_DEFAULT_STEM_INSTRUMENTS = ("bass", "other", "vocals") + + +class Mt3NotAvailableError(RuntimeError): + """Raised when the MT3 extra is not installed or the checkpoint is missing. + + Callers in ``analyze.py`` catch this (alongside ``Exception``) and emit a + ``[warn]`` line to stderr — MT3 failure must never block Phase 1 JSON. + """ + + +@dataclass +class Mt3Track: + """A single per-instrument MIDI track produced by MT3.""" + + instrument: str + midi_b64: str + note_count: int + pitch_range: tuple[int, int] + + +@dataclass +class Mt3Result: + """Top-level MT3 result for one ``transcribe()`` call.""" + + version: str + stems_used: list[str] + tracks: list[Mt3Track] = field(default_factory=list) + + def to_payload(self) -> dict: + """camelCase JSON projection (the analyze.py envelope is camelCase end-to-end).""" + return { + "version": self.version, + "stemsUsed": list(self.stems_used), + "tracks": [ + { + "instrument": track.instrument, + "midiB64": track.midi_b64, + "noteCount": int(track.note_count), + "pitchRange": [ + int(track.pitch_range[0]), + int(track.pitch_range[1]), + ], + } + for track in self.tracks + ], + } + + +def _resolve_checkpoint_dir() -> Path: + """Honor ``ASA_MT3_CHECKPOINT_DIR`` for advanced users; otherwise default.""" + override = os.getenv("ASA_MT3_CHECKPOINT_DIR", "").strip() + if override: + return Path(override).expanduser().resolve() + return DEFAULT_MT3_CHECKPOINT_DIR + + +def discover_stems_dir(stems: dict | None) -> Path | None: + """Recover the Demucs stems directory from an analyze.py ``stems`` dict. + + ``analyze_audio_io.separate_stems()`` returns a mapping like + ``{"drums": "/path/to/.../drums.wav", "bass": "...", ...}`` — every stem + file is written into the same parent directory. This helper recovers + that parent so ``transcribe()`` can discover per-stem files by canonical + filename via :func:`_resolve_sources`. + + Lives here (not in analyze.py) so the gate in analyze.py can stay + trivial and unit-testable behind a single import. Returns ``None`` when + ``stems`` is missing, empty, or contains no on-disk paths — callers + then fall back to running MT3 on the full mix. + """ + if not isinstance(stems, dict): + return None + for source_path in stems.values(): + if isinstance(source_path, str) and os.path.isfile(source_path): + return Path(source_path).parent + return None + + +def _resolve_sources( + audio_path: Path, + stems_dir: Path | None, + *, + instruments: Iterable[str] = _DEFAULT_STEM_INSTRUMENTS, +) -> list[tuple[str, Path]]: + """Stems-first, full-mix fallback. + + Discovers per-stem files inside ``stems_dir`` by the canonical Demucs + filenames (``bass.wav``, ``other.wav``, ``vocals.wav``). If ``stems_dir`` + is missing or contains none of the expected files, falls back to running + MT3 once on the full mix (allowed by the goal's Constraints). + """ + sources: list[tuple[str, Path]] = [] + if stems_dir is not None: + stems_dir = Path(stems_dir) + if stems_dir.is_dir(): + for stem_name in instruments: + for suffix in (".wav", ".flac"): + candidate = stems_dir / f"{stem_name}{suffix}" + if candidate.is_file(): + sources.append((stem_name, candidate)) + break + if not sources: + return [("full_mix", Path(audio_path))] + return sources + + +def transcribe(audio_path: Path, *, stems_dir: Path | None) -> Mt3Result: + """Run MT3 transcription against ``audio_path``, stems-first when possible. + + Args: + audio_path: Absolute path to the source audio (full mix). + stems_dir: Optional directory containing Demucs stems. When present, + MT3 runs once per discovered stem and emits one track per stem; + otherwise it falls back to a single full-mix pass. + + Returns: + ``Mt3Result`` with the pinned version string, the list of stem names + actually transcribed, and one ``Mt3Track`` per stem (each carrying a + base64-encoded MIDI blob). + + Raises: + Mt3NotAvailableError: when the ``[mt3]`` extra isn't installed or the + checkpoint directory is missing. Callers wrap this so the failure + surfaces as a ``[warn]`` rather than blocking Phase 1 JSON. + FileNotFoundError: when ``audio_path`` doesn't exist on disk. + """ + audio_path = Path(audio_path) + if not audio_path.is_file(): + raise FileNotFoundError(f"MT3 audio path missing: {audio_path}") + + # Lazy import — keeps JAX/t5x out of the import graph until MT3 is + # explicitly requested. Any ImportError surfaces as Mt3NotAvailableError + # so analyze.py's outer try/except can convert it to a [warn] line + # without leaking implementation detail. + try: + import librosa # already a base ASA dep, but kept inside the try + import note_seq + from mt3 import inference_model # type: ignore + except ImportError as exc: + raise Mt3NotAvailableError( + "MT3 backend not installed. Install the extra into a SEPARATE venv " + "(never the product venv) per apps/backend/requirements-mt3.txt; see " + "the apps/backend/mt3_transcription.py module docstring for the full " + "setup including the weight-download command." + ) from exc + + checkpoint_dir = _resolve_checkpoint_dir() + if not checkpoint_dir.is_dir(): + raise Mt3NotAvailableError( + f"MT3 checkpoint missing at {checkpoint_dir}. Download via: " + f"gsutil -m cp -r gs://mt3/checkpoints/mt3 {checkpoint_dir.parent}/" + ) + + sources = _resolve_sources(audio_path, stems_dir) + full_mix_fallback = ( + len(sources) == 1 and sources[0][0] == "full_mix" + ) + if full_mix_fallback: + print( + "[warn] MT3: running on full mix — Demucs stems not found at " + f"{stems_dir}; per-instrument separation will be weaker.", + file=sys.stderr, + ) + + # Build the inferencer once and reuse across stems — model load is the + # expensive step (~5-15s, multi-GB weights). MT3's InferenceModel handles + # the gin config + checkpoint restore internally. + try: + inferencer = inference_model.InferenceModel( + checkpoint_path=str(checkpoint_dir), + model_type="mt3", + ) + except Exception as exc: # noqa: BLE001 - surface as a typed init failure + raise Mt3NotAvailableError( + f"Failed to initialize MT3 InferenceModel from {checkpoint_dir}: {exc}" + ) from exc + + tracks: list[Mt3Track] = [] + stems_used: list[str] = [] + + for stem_name, source_path in sources: + try: + # MT3 was trained on 16 kHz mono audio. librosa handles resample + + # mono mixdown deterministically. + audio, _sr = librosa.load(str(source_path), sr=16000, mono=True) + + note_sequence = inferencer(audio) + + midi_buffer = BytesIO() + note_seq.note_sequence_to_pretty_midi(note_sequence).write(midi_buffer) + midi_b64 = base64.b64encode(midi_buffer.getvalue()).decode("ascii") + + pitches = [int(note.pitch) for note in note_sequence.notes] + pitch_range = ( + (min(pitches), max(pitches)) if pitches else (0, 0) + ) + + tracks.append( + Mt3Track( + instrument=stem_name, + midi_b64=midi_b64, + note_count=len(note_sequence.notes), + pitch_range=pitch_range, + ) + ) + stems_used.append(stem_name) + except Exception as exc: # noqa: BLE001 - per-stem failure is non-fatal + print( + f"[warn] MT3 stem {stem_name!r} failed: {exc}", + file=sys.stderr, + ) + # Continue to the next stem; partial results are valid output. + + return Mt3Result( + version=f"mt3-py-{MT3_MODULE_VERSION}+{MT3_CHECKPOINT_ID}", + stems_used=stems_used, + tracks=tracks, + ) + + +__all__ = [ + "DEFAULT_MT3_CHECKPOINT_DIR", + "MT3_CHECKPOINT_ID", + "MT3_MODULE_VERSION", + "Mt3NotAvailableError", + "Mt3Result", + "Mt3Track", + "discover_stems_dir", + "transcribe", +] diff --git a/apps/backend/prompts/phase2_system.txt b/apps/backend/prompts/phase2_system.txt index b7bfcf97..3a21c045 100644 --- a/apps/backend/prompts/phase2_system.txt +++ b/apps/backend/prompts/phase2_system.txt @@ -3,9 +3,10 @@ You are an expert Ableton Live 12 reconstruction producer. You receive: 1. AUTHORITATIVE_MEASUREMENT_RESULT_JSON 2. OPTIONAL_PITCH_NOTE_TRANSLATION_RESULT_JSON -3. LIVE_12_DEVICE_CATALOG_JSON -4. GROUNDING_METADATA -5. the audio file itself +3. OPTIONAL_MT3_TRANSCRIPTION_RESULT_JSON +4. LIVE_12_DEVICE_CATALOG_JSON +5. GROUNDING_METADATA +6. the audio file itself Your job is to turn the measured track facts into a practical Ableton Live rebuild blueprint. @@ -14,10 +15,11 @@ ABSOLUTE RULES 2. Never override measured BPM, key, loudness, duration, stereo metrics, or segment timing from audio perception. 3. Use the exact key string provided. Do not reinterpret it as a relative major/minor. 4. OPTIONAL_PITCH_NOTE_TRANSLATION_RESULT_JSON is best-effort note scaffolding only. Use it for note names, register hints, and pattern hints. Do not let it override measured timing, key, loudness, or structure. -5. LIVE_12_DEVICE_CATALOG_JSON is authoritative for device naming. Only emit devices and parameter labels that exist in that catalog. -6. Allowed device families are NATIVE and MAX_FOR_LIVE only. Never output third-party devices. -7. Do not output generic tutorials. Every important recommendation must be tied to this track’s measured values. -8. If a recommendation is uncertain, say why in confidenceNotes instead of pretending certainty. +5. OPTIONAL_MT3_TRANSCRIPTION_RESULT_JSON is best-effort polyphonic note scaffolding only — additive context, never authoritative. It is null unless the user opted into MT3 transcription and it succeeded; when null, ignore it entirely. When present, use it the same way as the pitch/note translation: for instrument-layer counts, polyphony hints, and per-track register hints. Never let it override measured BPM, key, loudness, duration, or structure, and never treat it as a vote against the OPTIONAL_PITCH_NOTE_TRANSLATION_RESULT_JSON note data — the two coexist. +6. LIVE_12_DEVICE_CATALOG_JSON is authoritative for device naming. Only emit devices and parameter labels that exist in that catalog. +7. Allowed device families are NATIVE and MAX_FOR_LIVE only. Never output third-party devices. +8. Do not output generic tutorials. Every important recommendation must be tied to this track’s measured values. +9. If a recommendation is uncertain, say why in confidenceNotes instead of pretending certainty. LIVE 12 OUTPUT POLICY - Use the exact device spelling from LIVE_12_DEVICE_CATALOG_JSON. diff --git a/apps/backend/requirements-mt3.txt b/apps/backend/requirements-mt3.txt new file mode 100644 index 00000000..884e44f2 --- /dev/null +++ b/apps/backend/requirements-mt3.txt @@ -0,0 +1,59 @@ +# Optional MT3 polyphonic-transcription extra. +# +# These deps are NOT part of the product baseline and must NOT be added to +# requirements.txt or installed into the product venv by default. The base +# ASA install runs without them; mt3_transcription.transcribe() lazy-imports +# everything below and raises Mt3NotAvailableError if any are missing. +# +# ASA's backend uses requirements-*.txt files in place of pyproject.toml +# extras (mirroring requirements-eval.txt for the beat-evaluation gate). +# This file is the project-idiom equivalent of a "[mt3] extra". +# +# Install into a SEPARATE venv — never the product venv. JAX/t5x and note_seq +# pull older numpy / protobuf transitively, while the product venv pins +# numpy==2.4.3 and protobuf==7.34.0 (see requirements.txt); JAX and PyTorch +# also want different CUDA stacks on some hosts. Installing this into the +# product venv risks breaking Essentia/torch on the request path. The +# lazy-import in mt3_transcription.transcribe() exists precisely so the +# product install never needs any of these: +# +# python3.11 -m venv apps/backend/venv-mt3 +# apps/backend/venv-mt3/bin/pip install -r apps/backend/requirements.txt \ +# -r apps/backend/requirements-mt3.txt +# +# Then download the checkpoint: +# +# mkdir -p apps/backend/models/mt3 +# gsutil -m cp -r gs://mt3/checkpoints/mt3 apps/backend/models/mt3/ +# +# Finally enable the gate for an analyze.py invocation: +# +# export ASA_ENABLE_MT3=1 +# +# MT3 itself is unmaintained upstream; we pin to a known-good commit rather +# than chasing main. note_seq and pretty_midi are stable. JAX + t5x are the +# fragile pieces — if upstream MT3 stops working with newer JAX, downgrade +# both together. See apps/backend/mt3_transcription.py for the JSON +# contract these deps satisfy. + +# MT3 itself. Upstream is unmaintained; the install command below pulls +# whatever main is at install time. Pinning to a specific commit is +# RECOMMENDED before sharing this file across machines — replace the URL +# with ``@`` once you've verified a working revision against +# the checkpoint at gs://mt3/checkpoints/mt3. No hash is pinned here +# because we can't verify upstream revisions from inside this repo. +mt3 @ git+https://github.com/magenta/mt3.git + +# Symbolic-music primitives MT3 emits. pretty_midi is used in the test +# suite to verify the base64-decoded MIDI parses cleanly. +note_seq>=0.0.5,<0.1.0 +pretty_midi>=0.2.10,<0.3.0 + +# JAX/t5x stack. MT3's gin configs are sensitive to t5x major version — +# stay on the 0.0.x series that the pinned MT3 commit was tested against. +jax[cpu]>=0.4.20,<0.5.0 +jaxlib>=0.4.20,<0.5.0 +flax>=0.7.0,<0.10.0 +t5x>=0.0.0,<0.1.0 +seqio>=0.0.18,<0.1.0 +gin-config>=0.5.0,<0.6.0 diff --git a/apps/backend/server.py b/apps/backend/server.py index bd8d91c8..5d8ad17c 100644 --- a/apps/backend/server.py +++ b/apps/backend/server.py @@ -33,6 +33,7 @@ from auth_context import AuthenticationRequiredError, UserContext, resolve_api_user_context from analysis_runtime import ( AnalysisRuntime, + UnsupportedMt3ModeError, UnsupportedPitchNoteBackendError, UnsupportedPitchNoteModeError, ) @@ -297,6 +298,7 @@ def _create_background_tasks( asyncio.create_task(_measurement_worker_loop()), asyncio.create_task(_pitch_note_worker_loop()), asyncio.create_task(_interpretation_worker_loop()), + asyncio.create_task(_mt3_worker_loop()), ] ) return tasks @@ -540,6 +542,31 @@ def _run_streamed_subprocess( } +def _coerce_mt3_mode(value: Any) -> str: + """Validate/normalize a route-form mt3_mode. + + The HTTP route declares ``mt3_mode: str = Form("off")``. When the route + function is invoked directly (not via HTTP dispatch — e.g. the unit tests + that call ``server.create_analysis_run(...)`` to bypass multipart parsing), + FastAPI leaves the default as its internal ``Form(...)`` sentinel object + rather than the string ``"off"``. That sentinel — and the empty string, + which is how an absent multipart field can arrive — normalize to the + conservative ``"off"`` default (the "absent unless requested" contract). + + A real, non-empty string is validated strictly: ``"off"`` / ``"enabled"`` + pass through; anything else raises :class:`UnsupportedMt3ModeError` so a + client typo like ``"enable"`` surfaces as ``400 MT3_MODE_UNSUPPORTED`` + rather than silently disabling MT3. This mirrors how ``pitch_note_mode`` is + rejected and keeps the typed error / catch / code machinery (and the + contract documented in ``analysisRunsClient.ts``) live rather than dead. + """ + if not isinstance(value, str) or value == "": + return "off" + if value in {"off", "enabled"}: + return value + raise UnsupportedMt3ModeError(value) + + async def _create_analysis_run_record( *, track: UploadFile, @@ -551,7 +578,9 @@ async def _create_analysis_run_record( interpretation_profile: str, interpretation_model: str | None, legacy_request_id: str | None = None, + mt3_mode: Any = "off", ) -> tuple[AnalysisRuntime, str]: + mt3_mode = _coerce_mt3_mode(mt3_mode) temp_path: str | None = None runtime = get_analysis_runtime() analysis_mode = _resolve_analysis_mode_value(analysis_mode) @@ -582,6 +611,7 @@ async def _create_analysis_run_record( interpretation_profile=interpretation_profile, interpretation_model=interpretation_model, legacy_request_id=legacy_request_id, + mt3_mode=mt3_mode, ) return runtime, created["runId"] @@ -597,6 +627,7 @@ async def _create_analysis_run_record_from_url( interpretation_profile: str, interpretation_model: str | None, legacy_request_id: str | None = None, + mt3_mode: Any = "off", ) -> tuple[AnalysisRuntime, str]: """Parallel of ``_create_analysis_run_record`` but for URL-fetched audio. @@ -605,6 +636,7 @@ async def _create_analysis_run_record_from_url( errors are surfaced unchanged to the caller, which is responsible for translating them into HTTP envelopes. """ + mt3_mode = _coerce_mt3_mode(mt3_mode) runtime = get_analysis_runtime() analysis_mode = _resolve_analysis_mode_value(analysis_mode) if interpretation_mode != "off": @@ -624,6 +656,7 @@ async def _create_analysis_run_record_from_url( interpretation_profile=interpretation_profile, interpretation_model=interpretation_model, legacy_request_id=legacy_request_id, + mt3_mode=mt3_mode, ) return runtime, created["runId"] @@ -670,6 +703,7 @@ async def _estimate_analysis_run( interpretation_mode: str, interpretation_profile: str, interpretation_model: str | None, + mt3_mode: Any = "off", run_separation_override: bool | None = None, run_transcribe_override: bool | None = None, ) -> JSONResponse: @@ -696,11 +730,13 @@ async def _estimate_analysis_run( if run_transcribe_override is None else run_transcribe_override ) + run_mt3 = _coerce_mt3_mode(mt3_mode) == "enabled" estimate = _build_backend_estimate( temp_path, run_separation, run_transcribe, analysis_mode=analysis_mode, + run_mt3=run_mt3, ) return JSONResponse( content={ @@ -1207,14 +1243,21 @@ def _execute_pitch_note_attempt( started_at = _current_time() run_id = str(attempt["runId"]) attempt_id = str(attempt["attemptId"]) - source_artifact = runtime.get_source_artifact(run_id) provenance: dict[str, Any] = { "schemaVersion": "pitch_note_translation.v1", "backendId": attempt["backendId"], "mode": attempt["mode"], } + # Bound before the try so the except handler can reference it even when + # get_source_artifact (now inside the try) is the call that raised. + source_artifact: dict[str, Any] | None = None stem_output_dir: str | None = None try: + # Resolve the source artifact INSIDE the try so a failure here + # terminalizes the attempt rather than leaving it stuck 'running' — which + # the interpretation-ordering gate would otherwise turn into an indefinite + # block on this run's interpretation. (Symmetry with _execute_mt3_attempt.) + source_artifact = runtime.get_source_artifact(run_id) source_local_path = runtime.require_local_artifact_path( source_artifact.get("path"), purpose="Source audio artifact for pitch/note translation", @@ -1320,7 +1363,8 @@ def _execute_pitch_note_attempt( provenance=provenance, diagnostics={ "backendDurationMs": round(_elapsed_ms(started_at, _current_time()), 2), - "sourceArtifactId": source_artifact["artifactId"], + # Null-safe: source_artifact is None if get_source_artifact itself raised. + "sourceArtifactId": (source_artifact or {}).get("artifactId"), "isolationMode": "subprocess", }, ) @@ -1329,6 +1373,235 @@ def _execute_pitch_note_attempt( shutil.rmtree(stem_output_dir, ignore_errors=True) +# MT3 stage error-classification markers. The MT3 module emits these +# verbatim in its Mt3NotAvailableError messages; matching on stderr is +# fragile but cheap and avoids round-tripping a structured error object +# through stdout. If mt3_transcription.py changes its phrasing, update +# both sides together. +_MT3_NOT_AVAILABLE_MARKERS = ( + "MT3 backend not installed", + "MT3 checkpoint missing", + "Failed to initialize MT3 InferenceModel", +) + + +def _execute_mt3_attempt( + runtime: AnalysisRuntime, + attempt: dict[str, Any], +) -> None: + """Run MT3 polyphonic transcription as a subprocess. + + Subprocess isolation matters even more here than for pitch_note: MT3 + loads multi-GB JAX/t5x model weights and a freshly-imported JAX + pollutes the parent's import graph in ways that interact badly with + the rest of the analyse server. Subprocess exit reclaims everything. + + MVP scope: runs MT3 on the full mix unless a previous stage (pitch_note) + has already persisted Demucs stems as artifacts. We do NOT invoke + Demucs from this stage today — the follow-up step adds bidirectional + stems handover (either stage caches stems for the other to consume). + """ + started_at = _current_time() + run_id = str(attempt["runId"]) + attempt_id = str(attempt["attemptId"]) + checkpoint_id = str(attempt.get("checkpointId") or "") + provenance: dict[str, Any] = { + "schemaVersion": "mt3_transcription.v1", + "checkpointId": checkpoint_id, + } + # Bound before the try so the except handlers can reference it even when + # get_source_artifact (now inside the try) is the call that raised. + source_artifact: dict[str, Any] | None = None + midi_tempdir: str | None = None + try: + # Resolve the source artifact INSIDE the try. If get_source_artifact + # raised here (e.g. a missing artifact row), the attempt would otherwise + # escape to _mt3_worker_loop stuck in 'running' — and with the + # interpretation-ordering gate that would block this run's interpretation + # indefinitely (until restart recovery). Terminalizing it on failure + # keeps the gate's no-deadlock guarantee airtight. + source_artifact = runtime.get_source_artifact(run_id) + source_local_path = runtime.require_local_artifact_path( + source_artifact.get("path"), + purpose="Source audio artifact for MT3 transcription", + ) + command = [ + "./venv/bin/python", "analyze.py", + source_local_path, + "--mt3-only", + "--yes", + ] + + # If pitch_note has already run, its stems are persisted as + # internal artifacts (kind = "stem_"). Reuse them by + # passing the common parent directory so the MT3 subprocess + # discovers them via mt3_transcription._resolve_sources. + existing_stems = runtime.get_internal_artifacts_by_kind(run_id, "stem_") + stem_paths_map = { + artifact["kind"].removeprefix("stem_"): str(local_path) + for artifact in existing_stems + if (local_path := runtime.resolve_artifact_local_path(artifact.get("path"))) is not None + and local_path.is_file() + } + if "bass" in stem_paths_map or "other" in stem_paths_map: + stem_dirs = {os.path.dirname(p) for p in stem_paths_map.values()} + if len(stem_dirs) == 1: + command.extend(["--stem-dir", stem_dirs.pop()]) + + # MT3 model load can take ~30s on first call; long tracks add + # several minutes of inference. 1800s gives generous headroom + # while still bounding the worker. + result = subprocess.run( + command, + capture_output=True, + text=True, + check=False, + timeout=1800, + ) + + if result.returncode != 0: + stderr_tail = result.stderr[-2000:] if result.stderr else "" + if any(marker in stderr_tail for marker in _MT3_NOT_AVAILABLE_MARKERS): + raise _Mt3UnavailableError( + f"MT3 backend unavailable: {stderr_tail[-500:]}" + ) + raise RuntimeError( + f"MT3 subprocess failed (exit {result.returncode}): " + f"{stderr_tail[-500:] if stderr_tail else 'no stderr'}" + ) + + mt3_payload = json.loads(result.stdout) + if not isinstance(mt3_payload, dict): + raise RuntimeError( + f"MT3 subprocess produced non-dict JSON output: {type(mt3_payload).__name__}" + ) + + # Swap each track's inline midiB64 for a persisted artifact ref. + # This is the load-bearing decision from the design: snapshot + # polls must stay small (KB), not balloon to MB per track. + tracks_in = mt3_payload.get("tracks") or [] + tracks_out: list[dict[str, Any]] = [] + midi_tempdir = tempfile.mkdtemp( + prefix="asa_mt3_midi_", + dir=str(runtime.runtime_dir), + ) + for track in tracks_in: + if not isinstance(track, dict): + continue + midi_b64 = track.get("midiB64") + if not isinstance(midi_b64, str) or not midi_b64: + # Track with no MIDI body — keep the metadata but skip + # the artifact step. This shouldn't happen in practice + # (the MT3 module emits empty-bytes base64 for empty + # tracks) but defends against malformed subprocess output. + tracks_out.append( + { + "instrument": str(track.get("instrument", "unknown")), + "midiArtifactId": None, + "midiSizeBytes": 0, + "noteCount": int(track.get("noteCount", 0)), + "pitchRange": list(track.get("pitchRange", [0, 0])), + } + ) + continue + instrument = str(track.get("instrument", "unknown")) + midi_bytes = base64.b64decode(midi_b64) + midi_path = os.path.join(midi_tempdir, f"{instrument}.mid") + with open(midi_path, "wb") as midi_file: + midi_file.write(midi_bytes) + artifact = runtime.record_artifact( + run_id, + kind=f"mt3_track_{instrument}", + source_path=midi_path, + filename=f"mt3_{instrument}.mid", + mime_type="audio/midi", + provenance={ + "schemaVersion": "artifact.v1", + "sourceArtifactId": source_artifact["artifactId"], + "generator": "mt3_transcription_subprocess", + "instrument": instrument, + "checkpointId": checkpoint_id, + }, + ) + tracks_out.append( + { + "instrument": instrument, + "midiArtifactId": str(artifact["artifactId"]), + "midiSizeBytes": int(artifact["sizeBytes"]), + "noteCount": int(track.get("noteCount", 0)), + "pitchRange": list(track.get("pitchRange", [0, 0])), + } + ) + + mt3_result = { + "version": str(mt3_payload.get("version") or ""), + "stemsUsed": list(mt3_payload.get("stemsUsed") or []), + "tracks": tracks_out, + } + + diagnostics = { + "backendDurationMs": round(_elapsed_ms(started_at, _current_time()), 2), + "stemSeparationUsed": bool(stem_paths_map), + "sourceArtifactId": source_artifact["artifactId"], + "isolationMode": "subprocess", + "trackCount": len(tracks_out), + } + # Surface the actual checkpoint string the subprocess reported so + # Phase 2 (when wired) can attribute notes to the real revision. + if isinstance(mt3_result["version"], str) and mt3_result["version"]: + provenance["resolvedCheckpointId"] = mt3_result["version"] + runtime.complete_mt3_attempt( + attempt_id, + result=mt3_result, + provenance=provenance, + diagnostics=diagnostics, + ) + except _Mt3UnavailableError as exc: + runtime.fail_mt3_attempt( + attempt_id, + error={ + "code": "MT3_NOT_AVAILABLE", + "message": str(exc), + "retryable": False, + "phase": "mt3_transcription", + }, + provenance=provenance, + diagnostics={ + "backendDurationMs": round(_elapsed_ms(started_at, _current_time()), 2), + "sourceArtifactId": source_artifact["artifactId"], + "isolationMode": "subprocess", + }, + ) + except Exception as exc: # noqa: BLE001 - catch-all per stage convention + runtime.fail_mt3_attempt( + attempt_id, + error={ + "code": "MT3_TRANSCRIPTION_FAILED", + "message": str(exc), + "retryable": True, + "phase": "mt3_transcription", + }, + provenance=provenance, + diagnostics={ + "backendDurationMs": round(_elapsed_ms(started_at, _current_time()), 2), + # Null-safe: source_artifact is None if get_source_artifact itself raised. + "sourceArtifactId": (source_artifact or {}).get("artifactId"), + "isolationMode": "subprocess", + }, + ) + finally: + if midi_tempdir is not None: + shutil.rmtree(midi_tempdir, ignore_errors=True) + + +class _Mt3UnavailableError(RuntimeError): + """Internal marker raised by _execute_mt3_attempt when stderr matches + the mt3_transcription Mt3NotAvailableError phrasing. Separate from the + typed exception in mt3_transcription so server.py doesn't need to + import the MT3 module (which would pull mt3_transcription's lazy + import graph into the parent process).""" + + def _run_interpretation_request( *, source_path: str, @@ -1340,6 +1613,7 @@ def _run_interpretation_request( grounding_metadata: dict[str, Any], model_name: str, request_id: str, + mt3_result: dict[str, Any] | None = None, ) -> dict[str, Any]: try: profile_config = _resolve_interpretation_profile_config(profile_id) @@ -1363,6 +1637,7 @@ def _run_interpretation_request( grounding_metadata=grounding_metadata, model_name=model_name, request_id=request_id, + mt3_result=mt3_result, ) @@ -1375,6 +1650,7 @@ def _run_combined_stem_summary_request( grounding_metadata: dict[str, Any], model_name: str, request_id: str, + mt3_result: dict[str, Any] | None = None, ) -> dict[str, Any]: stem_artifacts = runtime.get_internal_artifacts_by_kind(run_id, "stem_") usable_stems = [ @@ -1419,6 +1695,7 @@ def _run_combined_stem_summary_request( grounding_metadata=stem_grounding_metadata, model_name=model_name, request_id=f"{request_id}:{stem_kind}", + mt3_result=mt3_result, ) diagnostics_by_stem.append( { @@ -1478,6 +1755,7 @@ def _run_interpretation_request_with_profile_config( grounding_metadata: dict[str, Any], model_name: str, request_id: str, + mt3_result: dict[str, Any] | None = None, ) -> dict[str, Any]: request_started_at = _current_time() flags_used: list[str] = [] @@ -1520,6 +1798,7 @@ def _run_interpretation_request_with_profile_config( pitch_note_result=pitch_note_result, grounding_metadata=grounding_metadata, descriptor_hooks=descriptor_hooks, + mt3_result=mt3_result, ) client = _genai.Client( api_key=api_key, @@ -1617,7 +1896,11 @@ def _generate_files_api() -> Any: else [] ) citation_path_warnings = ( - _validate_phase2_citation_paths(interpretation_result, measurement_result) + _validate_phase2_citation_paths( + interpretation_result, + measurement_result, + mt3_result=mt3_result, + ) if profile_id == "producer_summary" and interpretation_result is not None else [] ) @@ -1696,12 +1979,20 @@ def _execute_interpretation_attempt( grounding = runtime.get_interpretation_grounding(run_id) measurement_result = grounding["measurementResult"] or {} pitch_note_result = grounding["pitchNoteResult"] + # MT3 polyphonic transcription, when present, is forwarded to Gemini as + # additive grounding (OPTIONAL_MT3_TRANSCRIPTION_RESULT_JSON). It is None + # unless an MT3 attempt completed for this run. PURPOSE.md invariant #1 — + # Phase 1 stays authoritative; MT3 never overrides measured values. + mt3_result = grounding.get("mt3Result") grounding_metadata = { "measurementIsAuthoritative": True, "pitchNoteTranslationIsBestEffort": True, + "mt3TranscriptionIsBestEffort": True, "measurementOutputId": grounding["measurementOutputId"], "pitchNoteAttemptId": grounding["pitchNoteAttemptId"], + "mt3AttemptId": grounding.get("mt3AttemptId"), "doNotPromotePitchNoteToMeasurement": True, + "doNotPromoteMt3ToMeasurement": True, "profileId": profile_id, } model_name = _coerce_string(attempt.get("modelName"), "gemini-2.5-flash") @@ -1714,6 +2005,7 @@ def _execute_interpretation_attempt( grounding_metadata=grounding_metadata, model_name=model_name, request_id=str(attempt["attemptId"]), + mt3_result=mt3_result, ) else: source_artifact = runtime.get_source_artifact(run_id) @@ -1731,6 +2023,7 @@ def _execute_interpretation_attempt( grounding_metadata=grounding_metadata, model_name=model_name, request_id=str(attempt["attemptId"]), + mt3_result=mt3_result, ) profile_config = _resolve_interpretation_profile_config(profile_id) provenance = { @@ -1857,6 +2150,32 @@ async def _interpretation_worker_loop() -> None: await asyncio.sleep(WORKER_IDLE_SECONDS) +async def _mt3_worker_loop() -> None: + """Peer of _pitch_note_worker_loop / _interpretation_worker_loop. + + Picks up queued MT3 attempts whose owning run's measurement stage has + completed, runs them through _execute_mt3_attempt, and idles between + polls. Wired into _create_background_tasks so it activates in both + local (`include_workers=True`) and hosted-worker process roles. + """ + while True: + try: + attempt = await asyncio.to_thread(get_analysis_runtime().reserve_next_mt3_attempt) + if attempt is None: + await asyncio.sleep(WORKER_IDLE_SECONDS) + continue + await asyncio.to_thread( + _execute_mt3_attempt, + get_analysis_runtime(), + attempt, + ) + except asyncio.CancelledError: + raise + except Exception as exc: + print(f"[warn] mt3 worker loop failed: {exc}", file=sys.stderr) + await asyncio.sleep(WORKER_IDLE_SECONDS) + + def _uploaded_file_size_bytes(track: UploadFile) -> int | None: file_obj = getattr(track, "file", None) @@ -1926,6 +2245,7 @@ async def create_analysis_run( interpretation_mode: str = Form("off"), interpretation_profile: str = Form("producer_summary"), interpretation_model: str | None = Form(None), + mt3_mode: str = Form("off"), x_asa_user_id: str | None = Header(None), x_asa_user_email: str | None = Header(None), ) -> JSONResponse: @@ -2016,6 +2336,7 @@ async def create_analysis_run( interpretation_mode=interpretation_mode, interpretation_profile=interpretation_profile, interpretation_model=interpretation_model, + mt3_mode=mt3_mode, ) else: runtime, run_id = await _create_analysis_run_record( @@ -2027,6 +2348,7 @@ async def create_analysis_run( interpretation_mode=interpretation_mode, interpretation_profile=interpretation_profile, interpretation_model=interpretation_model, + mt3_mode=mt3_mode, ) return JSONResponse( content=_normalize_run_snapshot( @@ -2046,6 +2368,20 @@ async def create_analysis_run( } }, ) + except UnsupportedMt3ModeError as exc: + # Explicit catch before the generic ValueError fall-through so + # the client gets a typed MT3_MODE_UNSUPPORTED code, not the + # generic INTERPRETATION_PROFILE_UNSUPPORTED fallback in + # _value_error_code. + return JSONResponse( + status_code=400, + content={ + "error": { + "code": "MT3_MODE_UNSUPPORTED", + "message": str(exc), + } + }, + ) except ValueError as exc: return JSONResponse( status_code=400, @@ -2080,6 +2416,11 @@ async def estimate_analysis_run( interpretation_mode: str = Form("off"), interpretation_profile: str = Form("producer_summary"), interpretation_model: str | None = Form(None), + # When mt3_mode == "enabled", _estimate_analysis_run forwards run_mt3 to + # build_analysis_estimate, which appends the MT3 polyphonic-transcription + # stage to the returned BackendEstimateStage list so the UI can show the + # added cost before the user commits to the run. + mt3_mode: str = Form("off"), x_asa_user_id: str | None = Header(None), x_asa_user_email: str | None = Header(None), ) -> JSONResponse: @@ -2096,6 +2437,7 @@ async def estimate_analysis_run( interpretation_mode=interpretation_mode, interpretation_profile=interpretation_profile, interpretation_model=interpretation_model, + mt3_mode=mt3_mode, ) except UploadTooLargeError as exc: return _canonical_upload_too_large_response(exc) @@ -2119,6 +2461,19 @@ async def estimate_analysis_run( } }, ) + except UnsupportedMt3ModeError as exc: + # Mirror the create-run route: an unrecognised mt3_mode (now raised by + # _coerce_mt3_mode) must surface as a typed MT3_MODE_UNSUPPORTED code, + # not the generic _value_error_code fall-through below. + return JSONResponse( + status_code=400, + content={ + "error": { + "code": "MT3_MODE_UNSUPPORTED", + "message": str(exc), + } + }, + ) except ValueError as exc: return JSONResponse( status_code=400, @@ -2854,6 +3209,63 @@ async def generate_spectral_enhancement( ) +@app.post("/api/analysis-runs/{run_id}/mt3-transcriptions") +async def create_mt3_transcription_attempt( + run_id: str, + x_asa_user_id: str | None = Header(None), + x_asa_user_email: str | None = Header(None), +) -> JSONResponse: + """Manually enqueue an MT3 attempt on an existing run. + + Mirror of :func:`create_pitch_note_translation_attempt`. Used when a + client wants to opt into MT3 *after* the run was created (e.g. the + initial POST /api/analysis-runs had ``mt3_mode='off'`` but the user + later decides they want a MIDI export). The measurement stage must be + completed first — MT3 reads stems and/or the source audio that + measurement persists as artifacts. + + Returns 202 with the post-enqueue snapshot. The worker (started by + ``_create_background_tasks``) will reserve and run the attempt. + + No request body / form fields today — MT3 has no configurable + backend or mode. If/when multiple checkpoints land, the route can + accept ``checkpoint_id`` without changing its URL. + """ + 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() + try: + runtime.get_run(run_id, owner_user_id=user_context.user_id) + if runtime.get_measurement_status(run_id) != "completed": + return JSONResponse( + status_code=409, + content={ + "error": { + "code": "MEASUREMENT_NOT_READY", + "message": "Measurement must complete before MT3 transcription can run.", + } + }, + ) + runtime.create_mt3_attempt( + run_id, + status="queued", + provenance={ + "schemaVersion": "mt3_transcription.v1", + "requestedViaApi": True, + }, + ) + return JSONResponse( + status_code=202, + content=_normalize_run_snapshot( + runtime.get_run(run_id, owner_user_id=user_context.user_id), + runtime, + ), + ) + except (KeyError, PermissionError): + return _run_not_found_response(run_id) + + @app.post("/api/analysis-runs/{run_id}/pitch-note-translations") async def create_pitch_note_translation_attempt( run_id: str, @@ -2977,6 +3389,7 @@ def _build_backend_estimate( run_transcribe: bool, *, analysis_mode: str = "full", + run_mt3: bool = False, ) -> dict[str, Any]: try: duration_seconds = get_audio_duration_seconds(audio_path) @@ -2990,12 +3403,14 @@ def _build_backend_estimate( run_separation, run_transcribe, run_standard=True, + run_mt3=run_mt3, ) else: raw_estimate = build_analysis_estimate( safe_duration, run_separation, run_transcribe, + run_mt3=run_mt3, ) raw_stages = raw_estimate.get("stages") stages = ( diff --git a/apps/backend/server_phase2.py b/apps/backend/server_phase2.py index 0258cc11..fa3ed798 100644 --- a/apps/backend/server_phase2.py +++ b/apps/backend/server_phase2.py @@ -793,6 +793,7 @@ def _build_phase2_prompt( pitch_note_result: dict[str, Any] | None, grounding_metadata: dict[str, Any], descriptor_hooks: dict[str, Any] | None = None, + mt3_result: dict[str, Any] | None = None, ) -> str: measurement_for_gemini = _normalize_measurement_result_for_gemini(measurement_result) sections = [ @@ -801,6 +802,16 @@ def _build_phase2_prompt( json.dumps(measurement_for_gemini, indent=2), "\n\nOPTIONAL_PITCH_NOTE_TRANSLATION_RESULT_JSON:\n", json.dumps(pitch_note_result, indent=2), + # MT3 polyphonic-transcription grounding. ADDITIVE only — Phase 1 + # measurements remain ground truth (PURPOSE.md invariant #1). The + # value is null unless the user opted into the MT3 stage AND the + # stage succeeded. When non-null, cite paths under + # `transcription.mt3.*` in phase1Fields just as you would for + # other paths; the response's `transcriptionDetail` (from + # torchcrepe) and `transcription.mt3` (from MT3) coexist and + # complement each other — never vote between them. + "\n\nOPTIONAL_MT3_TRANSCRIPTION_RESULT_JSON:\n", + json.dumps(mt3_result, indent=2), "\n\nLIVE_12_DEVICE_CATALOG_JSON:\n", json.dumps(LIVE12_DEVICE_CATALOG, indent=2), "\n\nGROUNDING_METADATA:\n", @@ -822,6 +833,7 @@ def _build_stem_summary_prompt( pitch_note_result: dict[str, Any] | None, grounding_metadata: dict[str, Any], descriptor_hooks: dict[str, Any], + mt3_result: dict[str, Any] | None = None, ) -> str: sections = [ STEM_SUMMARY_PROMPT_TEMPLATE.rstrip(), @@ -829,6 +841,8 @@ def _build_stem_summary_prompt( json.dumps(measurement_result, indent=2), "\n\nOPTIONAL_PITCH_NOTE_TRANSLATION_RESULT_JSON:\n", json.dumps(pitch_note_result, indent=2), + "\n\nOPTIONAL_MT3_TRANSCRIPTION_RESULT_JSON:\n", + json.dumps(mt3_result, indent=2), "\n\nMEASUREMENT_DERIVED_DESCRIPTOR_HOOKS:\n", json.dumps(descriptor_hooks, indent=2), "\n\nGROUNDING_METADATA:\n", @@ -2651,6 +2665,8 @@ def _validate_citation_paths_for_record( def _validate_phase2_citation_paths( phase2_result: dict[str, Any], measurement_result: dict[str, Any], + *, + mt3_result: dict[str, Any] | None = None, ) -> list[dict[str, Any]]: """Backend defense-in-depth mirror of the frontend's citation-existence check. @@ -2669,9 +2685,26 @@ def _validate_phase2_citation_paths( ``spectralCentroid`` -> ``spectralCentroidMean``, top-level and per-stem), so a raw payload would lack the very names Gemini cites and emit false positives. The frontend avoids this by walking the already-renamed ``Phase1Result``. + + F3: When ``mt3_result`` is provided, its paths are merged into the + allowed set under the ``transcription.mt3.*`` namespace — matching what + the frontend's ``projectPhase1FromRun`` does. This lets Gemini cite + e.g. ``transcription.mt3.noteCount`` or, for array-of-object fields, + ``transcription.mt3.tracks.instrument`` (the walker registers array-item + fields under ``prefix.key`` with no ``[i]`` index — same convention as + ``noveltyPeaks.time``) without the backend validator flagging it as an + invented path. """ normalized = _normalize_measurement_result_for_gemini(measurement_result) allowed = _collect_measurement_field_paths(normalized) + if isinstance(mt3_result, dict): + # Mirror the frontend projection: synthesize the same + # `transcription.mt3.*` shape that `projectPhase1FromRun` adds to + # Phase1Result, then collect its paths. This keeps the backend and + # frontend citation-validators agreeing on what counts as a valid + # MT3 citation path. + mt3_namespace = {"transcription": {"mt3": mt3_result}} + allowed = allowed | _collect_measurement_field_paths(mt3_namespace) warnings: list[dict[str, Any]] = [] for index, item in enumerate(phase2_result.get("mixAndMasterChain") or []): diff --git a/apps/backend/tests/test_analysis_runtime.py b/apps/backend/tests/test_analysis_runtime.py index ac312e06..ae1ae119 100644 --- a/apps/backend/tests/test_analysis_runtime.py +++ b/apps/backend/tests/test_analysis_runtime.py @@ -378,6 +378,17 @@ def test_stage_progress_updates_are_visible_in_stage_diagnostics(self) -> None: 1, ) + # reserve_next_interpretation_attempt now blocks while a pitch_note (or + # mt3) attempt is in-flight, so settle the reserved pitch_note attempt + # before reserving interpretation. This test asserts interpretation + # PROGRESS visibility, not the cross-stage gate (covered by the + # InterpretationGatingTests class below). + runtime.complete_pitch_note_attempt( + pitch_note_attempt_id, + result={"transcriptionMethod": "stub", "noteCount": 0, "notes": []}, + provenance={"backendId": "auto"}, + ) + interpretation_attempt = runtime.reserve_next_interpretation_attempt() self.assertIsNotNone(interpretation_attempt) interpretation_attempt_id = str(interpretation_attempt["attemptId"]) @@ -627,6 +638,225 @@ def test_interpretation_attempts_store_grounding_columns(self) -> None: self.assertEqual(row["grounded_measurement_output_id"], grounding["measurementOutputId"]) self.assertEqual(row["grounded_pitch_note_attempt_id"], pitch_note_attempt_id) + def test_interpretation_grounding_exposes_mt3_result(self) -> None: + """F3: get_interpretation_grounding surfaces a completed MT3 attempt so + _execute_interpretation_attempt can forward it to Gemini. When no MT3 + attempt exists the fields stay null / 'not_requested' (additive only).""" + runtime = self._runtime() + created = runtime.create_run( + filename="track.mp3", + content=b"fake-audio", + mime_type="audio/mpeg", + pitch_note_mode="off", + pitch_note_backend="auto", + interpretation_mode="async", + interpretation_profile="producer_summary", + interpretation_model="gemini-2.5-flash", + ) + run_id = created["runId"] + runtime.complete_measurement( + run_id, + payload={"bpm": 128, "durationSeconds": 60.0}, + provenance={"schemaVersion": "measurement.v1"}, + diagnostics={"backendDurationMs": 1000}, + ) + + # No MT3 attempt yet: grounding stays additive-null. + grounding_before = runtime.get_interpretation_grounding(run_id) + self.assertIsNone(grounding_before["mt3Result"]) + self.assertIsNone(grounding_before["mt3AttemptId"]) + self.assertEqual(grounding_before["mt3Status"], "not_requested") + + mt3_result = { + "version": "magenta-mt3-base", + "stemsUsed": ["bass", "other"], + "tracks": [ + { + "instrument": "bass", + "midiArtifactId": "artifact-1", + "midiSizeBytes": 256, + "noteCount": 42, + "pitchRange": [28, 52], + } + ], + } + mt3_attempt_id = runtime.create_mt3_attempt( + run_id, + status="completed", + result=mt3_result, + provenance={"resolvedCheckpointId": "magenta-mt3-base"}, + ) + + grounding_after = runtime.get_interpretation_grounding(run_id) + self.assertEqual(grounding_after["mt3Result"], mt3_result) + self.assertEqual(grounding_after["mt3AttemptId"], mt3_attempt_id) + self.assertEqual(grounding_after["mt3Status"], "completed") + + +class InterpretationGatingTests(unittest.TestCase): + """The interpretation stage must wait for the additive grounding peers + (mt3 + pitch_note) to settle before it runs, so Gemini Phase 2 actually + sees mt3Result / pitchNoteResult instead of None. reserve_next_interpretation_attempt + blocks while either peer is in-flight ('queued'/'running'); terminal states + (completed/failed/interrupted) unblock it. + """ + + def setUp(self) -> None: + self.temp_dir = tempfile.TemporaryDirectory(prefix="asa_interp_gate_test_") + + def tearDown(self) -> None: + self.temp_dir.cleanup() + + def _runtime(self): + from analysis_runtime import AnalysisRuntime + + return AnalysisRuntime(Path(self.temp_dir.name) / "runtime", max_pending_per_stage=4) + + def _run_ready_for_interpretation(self, runtime, *, mt3_mode="off", pitch_note_mode="off"): + """Create a run that requests interpretation (+ optionally mt3/pitch_note) + and complete its measurement, which enqueues the requested followups as + 'queued' via _enqueue_requested_followups (the real production path).""" + created = runtime.create_run( + filename="track.mp3", + content=b"fake-audio", + mime_type="audio/mpeg", + pitch_note_mode=pitch_note_mode, + pitch_note_backend="auto", + interpretation_mode="async", + interpretation_profile="producer_summary", + interpretation_model="gemini-2.5-flash", + mt3_mode=mt3_mode, + ) + run_id = created["runId"] + runtime.complete_measurement( + run_id, + payload={"bpm": 128, "durationSeconds": 60.0}, + provenance={"schemaVersion": "measurement.v1"}, + diagnostics={"backendDurationMs": 1000}, + ) + return run_id + + @staticmethod + def _mt3_result(): + return {"version": "magenta-mt3-base", "stemsUsed": [], "tracks": []} + + @staticmethod + def _pitch_note_result(): + return {"transcriptionMethod": "stub", "noteCount": 0, "notes": []} + + @staticmethod + def _stage_error(code): + return {"code": code, "message": "simulated", "retryable": False, "phase": "test"} + + def test_interpretation_blocked_while_mt3_in_flight(self) -> None: + # pitch_note off → isolate the mt3 guard. + runtime = self._runtime() + run_id = self._run_ready_for_interpretation(runtime, mt3_mode="enabled", pitch_note_mode="off") + # mt3 attempt is 'queued' → interpretation must not reserve. + self.assertIsNone(runtime.reserve_next_interpretation_attempt()) + mt3 = runtime.reserve_next_mt3_attempt() + self.assertIsNotNone(mt3) # mt3 now 'running' + self.assertIsNone(runtime.reserve_next_interpretation_attempt()) + + def test_interpretation_blocked_while_pitch_note_in_flight(self) -> None: + # mt3 off → isolate the pitch_note guard. + runtime = self._runtime() + run_id = self._run_ready_for_interpretation(runtime, mt3_mode="off", pitch_note_mode="stem_notes") + self.assertIsNone(runtime.reserve_next_interpretation_attempt()) + pn = runtime.reserve_next_pitch_note_attempt() + self.assertIsNotNone(pn) # pitch_note now 'running' + self.assertIsNone(runtime.reserve_next_interpretation_attempt()) + + def test_interpretation_waits_for_both_then_unblocks(self) -> None: + runtime = self._runtime() + run_id = self._run_ready_for_interpretation(runtime, mt3_mode="enabled", pitch_note_mode="stem_notes") + self.assertIsNone(runtime.reserve_next_interpretation_attempt()) # both queued + pn = runtime.reserve_next_pitch_note_attempt() + runtime.complete_pitch_note_attempt( + str(pn["attemptId"]), result=self._pitch_note_result(), provenance={"backendId": "auto"} + ) + # pitch_note done but mt3 still queued → still blocked. + self.assertIsNone(runtime.reserve_next_interpretation_attempt()) + mt3 = runtime.reserve_next_mt3_attempt() + runtime.complete_mt3_attempt( + str(mt3["attemptId"]), result=self._mt3_result(), provenance={} + ) + # both terminal → unblocks. + self.assertIsNotNone(runtime.reserve_next_interpretation_attempt()) + + def test_interpretation_reserved_after_grounding_failed(self) -> None: + runtime = self._runtime() + run_id = self._run_ready_for_interpretation(runtime, mt3_mode="enabled", pitch_note_mode="stem_notes") + pn = runtime.reserve_next_pitch_note_attempt() + runtime.fail_pitch_note_attempt(str(pn["attemptId"]), error=self._stage_error("PITCH_NOTE_TRANSLATION_FAILED")) + mt3 = runtime.reserve_next_mt3_attempt() + runtime.fail_mt3_attempt(str(mt3["attemptId"]), error=self._stage_error("MT3_TRANSCRIPTION_FAILED")) + # failed is terminal → interpretation unblocks (runs with grounding=None). + self.assertIsNotNone(runtime.reserve_next_interpretation_attempt()) + + def test_interpretation_reserved_immediately_without_grounding_stages(self) -> None: + # Regression guard: a run requesting neither grounding stage must not block. + runtime = self._runtime() + run_id = self._run_ready_for_interpretation(runtime, mt3_mode="off", pitch_note_mode="off") + self.assertIsNotNone(runtime.reserve_next_interpretation_attempt()) + + def test_interpretation_unblocks_after_recovery_clears_stale_running_grounding(self) -> None: + # A crash mid-grounding leaves rows 'running'; restart recovery interrupts + # them so interpretation can never deadlock forever. + runtime = self._runtime() + run_id = self._run_ready_for_interpretation(runtime, mt3_mode="enabled", pitch_note_mode="stem_notes") + runtime.reserve_next_pitch_note_attempt() # → running + runtime.reserve_next_mt3_attempt() # → running + self.assertIsNone(runtime.reserve_next_interpretation_attempt()) + runtime.recover_incomplete_attempts() # running → interrupted + self.assertIsNotNone(runtime.reserve_next_interpretation_attempt()) + + def test_enqueue_creates_interpretation_attempt_last(self) -> None: + # Regression guard for the enqueue-ordering fix: interpretation must be + # created AFTER mt3 and pitch_note so it can never be reserved in the + # window before a grounding row commits. Record the create order through + # the real _enqueue_requested_followups path (driven by complete_measurement). + runtime = self._runtime() + created = runtime.create_run( + filename="track.mp3", + content=b"fake-audio", + mime_type="audio/mpeg", + pitch_note_mode="stem_notes", + pitch_note_backend="auto", + interpretation_mode="async", + interpretation_profile="producer_summary", + interpretation_model="gemini-2.5-flash", + mt3_mode="enabled", + ) + run_id = created["runId"] + + call_order: list[str] = [] + originals = { + "pitch_note": runtime.create_pitch_note_attempt, + "mt3": runtime.create_mt3_attempt, + "interpretation": runtime.create_interpretation_attempt, + } + + def _recorder(name): + def _wrapped(*args, **kwargs): + call_order.append(name) + return originals[name](*args, **kwargs) + + return _wrapped + + runtime.create_pitch_note_attempt = _recorder("pitch_note") + runtime.create_mt3_attempt = _recorder("mt3") + runtime.create_interpretation_attempt = _recorder("interpretation") + + runtime.complete_measurement( + run_id, + payload={"bpm": 128, "durationSeconds": 60.0}, + provenance={"schemaVersion": "measurement.v1"}, + diagnostics={"backendDurationMs": 1000}, + ) + + self.assertEqual(call_order, ["pitch_note", "mt3", "interpretation"]) + class SpectralArtifactSnapshotTests(unittest.TestCase): """Verifies the STFT-spectrogram sampleRate provenance is exposed on the diff --git a/apps/backend/tests/test_analyze.py b/apps/backend/tests/test_analyze.py index d798b159..f45522ea 100644 --- a/apps/backend/tests/test_analyze.py +++ b/apps/backend/tests/test_analyze.py @@ -23,6 +23,19 @@ } # Top-level keys emitted by both full and fast modes — the shared output contract. +# +# Deliberately NOT in this set: ``transcription``. That namespace is emitted +# by the *direct CLI* path of MT3 polyphonic transcription +# (``mt3_transcription.py``) when ``ASA_ENABLE_MT3=1`` is set in the +# environment. The production path goes through the staged runtime +# (``analysis_runtime.py``'s ``mt3`` stage; route +# ``POST /api/analysis-runs/{id}/mt3-transcriptions``) and lives on +# ``snapshot.stages.mt3``, not the analyze.py JSON envelope. +# +# Fast mode's test enforces exact equality against this set (see +# ``test_output_schema_matches_full_mode``), so adding ``transcription`` +# here would force the field to always appear in CLI output and break the +# "absent when off" contract. See JSON_SCHEMA.md "Optional MT3 Namespace". EXPECTED_TOP_LEVEL_KEYS = { "bpm", "bpmConfidence", "bpmPercival", "bpmAgreement", "bpmDoubletime", "bpmSource", "bpmRawOriginal", @@ -2713,5 +2726,87 @@ def test_chord_templates_25_has_correct_shape_and_triad_masks(self) -> None: np.testing.assert_allclose(n_row, np.full(12, 1.0 / 12.0), atol=1e-9) +class BuildAnalysisEstimateMt3Tests(unittest.TestCase): + """Unit tests for the MT3 cost branch in build_analysis_estimate. + + The estimate endpoint surfaces these stage entries to the frontend so + users see a realistic time-range before opting into the MT3 stage. + """ + + def _build(self, duration_seconds: float, *, run_mt3: bool) -> dict: + from analyze_estimate import build_analysis_estimate + return build_analysis_estimate( + duration_seconds, + run_separation=False, + run_transcribe=False, + run_mt3=run_mt3, + ) + + def test_mt3_off_does_not_emit_mt3_stage(self) -> None: + estimate = self._build(240.0, run_mt3=False) + keys = [stage["key"] for stage in estimate["stages"]] + self.assertNotIn( + "mt3_transcription", + keys, + "mt3_transcription must be absent when run_mt3=False", + ) + + def test_mt3_on_emits_mt3_stage_with_documented_shape(self) -> None: + estimate = self._build(240.0, run_mt3=True) + mt3_stages = [ + stage for stage in estimate["stages"] if stage["key"] == "mt3_transcription" + ] + self.assertEqual(len(mt3_stages), 1) + mt3_stage = mt3_stages[0] + self.assertEqual(mt3_stage["label"], "MT3 polyphonic transcription") + self.assertIn("seconds", mt3_stage) + self.assertIn("min", mt3_stage["seconds"]) + self.assertIn("max", mt3_stage["seconds"]) + # Both bounds are non-negative integers and max >= min. + self.assertIsInstance(mt3_stage["seconds"]["min"], int) + self.assertIsInstance(mt3_stage["seconds"]["max"], int) + self.assertGreaterEqual(mt3_stage["seconds"]["min"], 0) + self.assertGreaterEqual(mt3_stage["seconds"]["max"], mt3_stage["seconds"]["min"]) + + def test_mt3_overhead_floor_dominates_short_clips(self) -> None: + """A 10s clip should hit the overhead floor (60s min, 180s max), + not the per-second ratio. Documents the cost model's high-fixed-cost + property — JAX/t5x warmup is the bottleneck for tiny clips.""" + estimate = self._build(10.0, run_mt3=True) + mt3_stages = [ + stage for stage in estimate["stages"] if stage["key"] == "mt3_transcription" + ] + self.assertEqual(len(mt3_stages), 1) + seconds = mt3_stages[0]["seconds"] + # 10s * 0.2 = 2 → overridden by 60s floor. + self.assertGreaterEqual(seconds["min"], 60) + # 10s * 0.8 = 8 → overridden by 180s floor. + self.assertGreaterEqual(seconds["max"], 180) + + def test_mt3_stage_scales_with_duration_on_long_clips(self) -> None: + """A 5-minute clip should exceed the overhead floors via the + per-second ratio — documents the per-second growth.""" + # 300s * 0.2 = 60 (== floor); 300s * 0.8 = 240 (> 180 floor). + estimate = self._build(300.0, run_mt3=True) + mt3_stages = [ + stage for stage in estimate["stages"] if stage["key"] == "mt3_transcription" + ] + seconds = mt3_stages[0]["seconds"] + self.assertGreater(seconds["max"], 180) + + def test_mt3_stage_contributes_to_total_seconds(self) -> None: + """Whatever the MT3 stage costs, it must roll up into the total.""" + without_mt3 = self._build(240.0, run_mt3=False) + with_mt3 = self._build(240.0, run_mt3=True) + self.assertGreater( + with_mt3["totalSeconds"]["min"], + without_mt3["totalSeconds"]["min"], + ) + self.assertGreater( + with_mt3["totalSeconds"]["max"], + without_mt3["totalSeconds"]["max"], + ) + + if __name__ == "__main__": unittest.main() diff --git a/apps/backend/tests/test_cleanup.py b/apps/backend/tests/test_cleanup.py index 4cab631d..cff43a72 100644 --- a/apps/backend/tests/test_cleanup.py +++ b/apps/backend/tests/test_cleanup.py @@ -156,7 +156,11 @@ def test_startup_schedules_recurring_cleanup_task_and_no_file_cache(self) -> Non runtime.recover_incomplete_attempts.assert_called_once_with() cleanup_loop.assert_called_once_with(runtime.runtime_dir) - self.assertEqual(create_task_mock.call_count, 4) + # 5 tasks: cache eviction, measurement worker, pitch_note worker, + # interpretation worker, mt3 worker. (The mt3 worker was added when + # MT3 polyphonic transcription became its own staged stage — + # see analysis_runtime.py and server.py::_mt3_worker_loop.) + self.assertEqual(create_task_mock.call_count, 5) info_mock.assert_called_once_with( "Upload limits configured: raw_audio_limit_bytes=%s edge_request_limit_bytes=%s", server.upload_limits.MAX_UPLOAD_SIZE_BYTES, diff --git a/apps/backend/tests/test_mt3_transcription.py b/apps/backend/tests/test_mt3_transcription.py new file mode 100644 index 00000000..4619913c --- /dev/null +++ b/apps/backend/tests/test_mt3_transcription.py @@ -0,0 +1,494 @@ +"""Tests for the optional MT3 polyphonic-transcription stage. + +Coverage: + 1. Pure unit tests of ``Mt3Result.to_payload()`` (the camelCase + serialization boundary) and ``_resolve_sources()`` (stems-first + + full-mix fallback). These are sub-second and always run. + 2. Direct ``transcribe()`` call: when the MT3 extra is not installed in + the test venv, the call must raise the typed ``Mt3NotAvailableError`` + rather than leak an ``ImportError``. Auto-skips if MT3 happens to be + installed locally. + 3. Subprocess tests against ``analyze.py``: verify the ``ASA_ENABLE_MT3`` + gate, the "transcription key absent when flag off" contract, and the + "MT3 failure never blocks Phase 1" contract. These run analyze.py for + real so they exercise the actual wiring, not a mock. + 4. Slow integration test (gated on ``RUN_SLOW_TESTS=1``): real MT3 + inference against a synthetic chord fixture, asserting non-empty + tracks and ``pretty_midi``-parseable MIDI. ASA's backend uses + ``unittest`` rather than ``pytest``; ``RUN_SLOW_TESTS=1`` plays the + role of ``@pytest.mark.slow`` selection. The MT3 model download is + multi-GB and is never run in CI. +""" + +from __future__ import annotations + +import base64 +import json +import os +import subprocess +import sys +import tempfile +import unittest +import wave +from io import BytesIO +from pathlib import Path + +import numpy as np + +# Import the module under test directly. mt3_transcription.py keeps all +# heavy JAX/t5x imports inside transcribe(); importing the module itself +# is stdlib + numpy only, so it's safe to do at module load. +_BACKEND_DIR = Path(__file__).resolve().parent.parent +if str(_BACKEND_DIR) not in sys.path: + sys.path.insert(0, str(_BACKEND_DIR)) + +from mt3_transcription import ( # noqa: E402 + MT3_CHECKPOINT_ID, + Mt3NotAvailableError, + Mt3Result, + Mt3Track, + _resolve_sources, + discover_stems_dir, + transcribe, +) + + +def _write_silent_wav(path: Path, sample_rate: int = 22_050, duration_seconds: float = 1.0) -> None: + """Write a tiny mono silent WAV. Used as a placeholder fixture for tests + that only need a path-on-disk audio file.""" + total_samples = int(sample_rate * duration_seconds) + pcm = np.zeros(total_samples, dtype=np.int16) + with wave.open(str(path), "wb") as wav_file: + wav_file.setnchannels(1) + wav_file.setsampwidth(2) + wav_file.setframerate(sample_rate) + wav_file.writeframes(pcm.tobytes()) + + +def _write_chord_fixture( + path: Path, + *, + sample_rate: int = 22_050, + duration_seconds: float = 10.0, +) -> None: + """Synthesize a polyphonic fixture for the slow MT3 integration test. + + A sustained C-major triad (C4, E4, G4) with a slow 0.5 Hz amplitude + envelope. Real-world enough that MT3 has clean note onsets to + transcribe, simple enough to stay deterministic across reruns. + """ + total = int(sample_rate * duration_seconds) + t = np.arange(total, dtype=np.float32) / sample_rate + chord = ( + 0.25 * np.sin(2 * np.pi * 261.63 * t) # C4 + + 0.25 * np.sin(2 * np.pi * 329.63 * t) # E4 + + 0.25 * np.sin(2 * np.pi * 392.00 * t) # G4 + ) + envelope = np.clip(0.5 + 0.5 * np.sin(2 * np.pi * 0.5 * t), 0.0, 1.0).astype(np.float32) + signal = (chord * envelope).astype(np.float32) + stereo = np.stack([signal, signal], axis=1) + pcm = np.clip(stereo, -1.0, 1.0) + pcm = (pcm * 32767.0).astype(np.int16) + with wave.open(str(path), "wb") as wav_file: + wav_file.setnchannels(2) + wav_file.setsampwidth(2) + wav_file.setframerate(sample_rate) + wav_file.writeframes(pcm.tobytes()) + + +class Mt3ResultShapeTests(unittest.TestCase): + """to_payload() emits the documented camelCase contract (tripwire #3).""" + + def test_to_payload_serializes_with_camelcase_keys(self) -> None: + result = Mt3Result( + version="mt3-py-0.1.0+test-checkpoint@0001-01-01", + stems_used=["bass", "other"], + tracks=[ + Mt3Track( + instrument="bass", + midi_b64="TVRoZA==", # MThd header bytes + note_count=12, + pitch_range=(36, 60), + ), + Mt3Track( + instrument="other", + midi_b64="TVRoZA==", + note_count=24, + pitch_range=(48, 84), + ), + ], + ) + payload = result.to_payload() + self.assertEqual( + set(payload.keys()), + {"version", "stemsUsed", "tracks"}, + "Top-level keys must be camelCase (analyze.py emits camelCase " + "directly; no conversion layer to src/types/measurement.ts).", + ) + self.assertEqual(payload["version"], "mt3-py-0.1.0+test-checkpoint@0001-01-01") + self.assertEqual(payload["stemsUsed"], ["bass", "other"]) + self.assertEqual(len(payload["tracks"]), 2) + first = payload["tracks"][0] + self.assertEqual( + set(first.keys()), + {"instrument", "midiB64", "noteCount", "pitchRange"}, + "Track entries must use camelCase keys.", + ) + self.assertEqual(first["instrument"], "bass") + self.assertEqual(first["midiB64"], "TVRoZA==") + self.assertEqual(first["noteCount"], 12) + self.assertEqual(first["pitchRange"], [36, 60]) + + def test_to_payload_handles_empty_result(self) -> None: + """An MT3 run that produces no usable notes still emits a valid + envelope — the wiring in analyze.py treats this as success.""" + result = Mt3Result( + version="mt3-py-0.1.0+test", + stems_used=[], + tracks=[], + ) + payload = result.to_payload() + self.assertEqual(payload["stemsUsed"], []) + self.assertEqual(payload["tracks"], []) + + def test_pinned_checkpoint_id_has_documented_format(self) -> None: + """Phase 2 reads MT3_CHECKPOINT_ID verbatim. The base identifier is + always present; an ``@`` suffix is optional but its + shape is constrained when operators add one (see the module docstring + on MT3_CHECKPOINT_ID). Don't relax this regex without updating the + module-level rationale comment.""" + self.assertRegex( + MT3_CHECKPOINT_ID, + r"^magenta-mt3-[a-z]+(@[A-Za-z0-9\-]+)?$", + ) + + +class Mt3SourceResolutionTests(unittest.TestCase): + """_resolve_sources() falls back to full-mix when stems are missing.""" + + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory(prefix="mt3_sources_") + self.addCleanup(self._tmp.cleanup) + self.audio = Path(self._tmp.name) / "mix.wav" + _write_silent_wav(self.audio, duration_seconds=0.5) + + def test_returns_full_mix_when_stems_dir_is_none(self) -> None: + sources = _resolve_sources(self.audio, None) + self.assertEqual(sources, [("full_mix", self.audio)]) + + def test_returns_full_mix_when_stems_dir_does_not_exist(self) -> None: + sources = _resolve_sources(self.audio, Path(self._tmp.name) / "nope") + self.assertEqual(sources, [("full_mix", self.audio)]) + + def test_returns_full_mix_when_stems_dir_is_empty(self) -> None: + empty = Path(self._tmp.name) / "stems_empty" + empty.mkdir() + sources = _resolve_sources(self.audio, empty) + self.assertEqual(sources, [("full_mix", self.audio)]) + + def test_discovers_canonical_demucs_stems_in_documented_order(self) -> None: + stems_dir = Path(self._tmp.name) / "stems" + stems_dir.mkdir() + for name in ("bass", "other", "vocals"): + (stems_dir / f"{name}.wav").write_bytes(b"\x00") + sources = _resolve_sources(self.audio, stems_dir) + # The order follows _DEFAULT_STEM_INSTRUMENTS so MT3 gets the same + # iteration order across runs — load-bearing for stems_used in the + # JSON envelope. + self.assertEqual([name for name, _ in sources], ["bass", "other", "vocals"]) + + def test_prefers_wav_over_flac_when_both_present(self) -> None: + stems_dir = Path(self._tmp.name) / "stems" + stems_dir.mkdir() + (stems_dir / "bass.wav").write_bytes(b"\x00") + (stems_dir / "bass.flac").write_bytes(b"\x00") + sources = _resolve_sources(self.audio, stems_dir) + self.assertEqual(sources, [("bass", stems_dir / "bass.wav")]) + + def test_skips_drums_stem_by_default(self) -> None: + """Drums are excluded by default (Phase 1 already covers them via + kickDetail/snareDetail/hihatDetail and MT3's drum head is weaker + than purpose-built drum trackers).""" + stems_dir = Path(self._tmp.name) / "stems" + stems_dir.mkdir() + (stems_dir / "drums.wav").write_bytes(b"\x00") + (stems_dir / "bass.wav").write_bytes(b"\x00") + sources = _resolve_sources(self.audio, stems_dir) + names = [name for name, _ in sources] + self.assertNotIn("drums", names) + self.assertIn("bass", names) + + +class Mt3StemsDirDiscoveryTests(unittest.TestCase): + """discover_stems_dir() recovers the Demucs parent directory. + + This is the helper called by analyze.py's gate to translate the + ``stems`` dict (which analyze_audio_io.separate_stems writes) into the + parent directory ``transcribe()`` expects. Unit-tested separately + because the subprocess pipeline test runs without ``--separate`` and + so never exercises this branch. + """ + + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory(prefix="mt3_discovery_") + self.addCleanup(self._tmp.cleanup) + self.stems_root = Path(self._tmp.name) / "stems" + self.stems_root.mkdir() + # Write the four Demucs canonical stems as silent WAVs. + self.stem_files: dict[str, Path] = {} + for name in ("drums", "bass", "other", "vocals"): + stem_path = self.stems_root / f"{name}.wav" + _write_silent_wav(stem_path, duration_seconds=0.1) + self.stem_files[name] = stem_path + + def test_recovers_parent_from_full_demucs_dict(self) -> None: + stems_dict = {name: str(path) for name, path in self.stem_files.items()} + self.assertEqual(discover_stems_dir(stems_dict), self.stems_root) + + def test_returns_none_for_none_input(self) -> None: + self.assertIsNone(discover_stems_dir(None)) + + def test_returns_none_for_empty_dict(self) -> None: + self.assertIsNone(discover_stems_dir({})) + + def test_returns_none_for_non_dict_input(self) -> None: + # Defensive — analyze.py passes whatever ``stems`` happens to be. + self.assertIsNone(discover_stems_dir([])) # type: ignore[arg-type] + self.assertIsNone(discover_stems_dir("not a dict")) # type: ignore[arg-type] + + def test_returns_none_when_paths_do_not_exist(self) -> None: + stems_dict = { + "drums": str(self.stems_root / "missing-drums.wav"), + "bass": str(self.stems_root / "missing-bass.wav"), + } + self.assertIsNone(discover_stems_dir(stems_dict)) + + def test_returns_parent_of_first_existing_path(self) -> None: + """First on-disk path wins — Demucs writes all stems to one dir, + so any one of them recovers the directory.""" + stems_dict = { + "drums": str(self.stems_root / "missing.wav"), + "bass": str(self.stem_files["bass"]), # exists + } + self.assertEqual(discover_stems_dir(stems_dict), self.stems_root) + + +def _is_mt3_extra_installed() -> bool: + try: + import mt3 # type: ignore # noqa: F401 + import note_seq # noqa: F401 + import librosa # noqa: F401 + except ImportError: + return False + return True + + +class Mt3NotAvailableTests(unittest.TestCase): + """When the [mt3] extra isn't installed, transcribe() raises a typed + Mt3NotAvailableError. analyze.py relies on this contract — it catches + Mt3NotAvailableError (alongside generic Exception) and surfaces it as a + [warn] line without blocking the Phase 1 JSON.""" + + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory(prefix="mt3_unavailable_") + self.addCleanup(self._tmp.cleanup) + self.audio = Path(self._tmp.name) / "mix.wav" + _write_silent_wav(self.audio, duration_seconds=0.5) + + @unittest.skipIf( + _is_mt3_extra_installed(), + "MT3 extra is installed locally; this test exercises the missing-deps path.", + ) + def test_raises_mt3_not_available_when_extra_missing(self) -> None: + with self.assertRaises(Mt3NotAvailableError): + transcribe(self.audio, stems_dir=None) + + def test_raises_file_not_found_for_missing_audio(self) -> None: + with self.assertRaises(FileNotFoundError): + transcribe(Path(self._tmp.name) / "nonexistent.wav", stems_dir=None) + + +class AnalyzePipelineGatingTests(unittest.TestCase): + """End-to-end: analyze.py honors the ASA_ENABLE_MT3 gate. + + These are subprocess tests that exercise the actual wiring in + apps/backend/analyze.py. They don't need the MT3 extra installed — + the contract under test is that the `transcription` JSON key is + absent (not null) by default, absent when MT3 fails, and absent in + fast mode regardless of the flag. + """ + + SAMPLE_RATE = 22_050 + DURATION = 4.0 + + @classmethod + def setUpClass(cls) -> None: + cls.backend_dir = Path(__file__).resolve().parent.parent + cls.analyze_path = cls.backend_dir / "analyze.py" + cls._tmp = tempfile.TemporaryDirectory(prefix="mt3_pipeline_") + cls.fixture = Path(cls._tmp.name) / "fixture.wav" + _write_chord_fixture( + cls.fixture, + sample_rate=cls.SAMPLE_RATE, + duration_seconds=cls.DURATION, + ) + + @classmethod + def tearDownClass(cls) -> None: + cls._tmp.cleanup() + + @classmethod + def _run_analyze(cls, *, env_overrides: dict[str, str], extra_args: list[str]) -> dict: + env = os.environ.copy() + # Strip any pre-existing flag so tests are hermetic. Override after. + env.pop("ASA_ENABLE_MT3", None) + env.update(env_overrides) + + venv_python = cls.backend_dir / "venv" / "bin" / "python" + python_exe = str(venv_python) if venv_python.is_file() else sys.executable + argv = [python_exe, str(cls.analyze_path), str(cls.fixture), "--yes", *extra_args] + + result = subprocess.run( + argv, + capture_output=True, + text=True, + env=env, + timeout=300, + ) + if result.returncode != 0: + raise AssertionError( + f"analyze.py exited {result.returncode}.\n" + f"argv: {argv}\n" + f"stdout (first 800):\n{result.stdout[:800]}\n" + f"stderr (first 800):\n{result.stderr[:800]}" + ) + try: + return json.loads(result.stdout) + except json.JSONDecodeError as exc: + raise AssertionError( + "analyze.py did not emit valid JSON.\n" + f"stdout (first 800):\n{result.stdout[:800]}\n" + f"stderr (first 800):\n{result.stderr[:800]}" + ) from exc + + def test_fast_mode_omits_transcription_when_flag_off(self) -> None: + """Baseline default: no flag, no MT3 namespace. Uses --fast for speed + because fast mode runs the same JSON-emission path that would be + affected by an accidental ``result["transcription"] = None``.""" + payload = self._run_analyze(env_overrides={}, extra_args=["--fast"]) + self.assertNotIn( + "transcription", + payload, + "transcription key must be ABSENT (not null) when ASA_ENABLE_MT3 " + "is unset — see JSON_SCHEMA.md 'Optional MT3 Namespace'.", + ) + + def test_fast_mode_omits_transcription_when_flag_on(self) -> None: + """Fast mode bypasses the MT3 hook entirely (analyze_fast.py builds + its own output dict). The flag has no effect here — load-bearing + defense-in-depth.""" + payload = self._run_analyze( + env_overrides={"ASA_ENABLE_MT3": "1"}, + extra_args=["--fast"], + ) + self.assertNotIn("transcription", payload) + # Sanity-check Phase 1 still emits: fast mode should never crash on + # the env var. + self.assertIn("bpm", payload) + + def test_full_mode_omits_transcription_and_keeps_phase1_when_deps_missing(self) -> None: + """Load-bearing 'never blocks Phase 1' contract: with the flag on but + no MT3 deps installed, the hook catches Mt3NotAvailableError, logs a + [warn], and analyze.py still emits a complete Phase 1 JSON without a + transcription key.""" + if _is_mt3_extra_installed(): + self.skipTest( + "MT3 extra is installed locally; this test exercises the " + "missing-deps failure path. Run the slow integration test " + "(RUN_SLOW_TESTS=1) instead to exercise real inference." + ) + payload = self._run_analyze( + env_overrides={"ASA_ENABLE_MT3": "1"}, + extra_args=[], + ) + self.assertIn( + "bpm", + payload, + "Phase 1 must still emit when MT3 fails — never block on optional stages.", + ) + self.assertNotIn( + "transcription", + payload, + "transcription key must be absent when MT3 deps are missing — " + "the failure path is logged as [warn], never as a null key.", + ) + + +@unittest.skipUnless( + os.environ.get("RUN_SLOW_TESTS") == "1", + "Set RUN_SLOW_TESTS=1 to run the real MT3 integration test. Requires " + "the [mt3] extra installed and the checkpoint downloaded — see " + "apps/backend/mt3_transcription.py and apps/backend/requirements-mt3.txt. " + "ASA's backend uses unittest, not pytest; this env-var skip plays the role " + "of @pytest.mark.slow in the goal text.", +) +class Mt3RealInferenceTests(unittest.TestCase): + """Live MT3 integration. Opt-in via RUN_SLOW_TESTS=1. Never runs in CI. + + Asserts the three concrete deliverables from the goal: + 1. Non-empty ``tracks``. + 2. Each ``midiB64`` parses cleanly via ``pretty_midi.PrettyMIDI``. + 3. The reported ``noteCount`` per track matches the parsed MIDI. + """ + + @classmethod + def setUpClass(cls) -> None: + cls._tmp = tempfile.TemporaryDirectory(prefix="mt3_slow_") + cls.fixture = Path(cls._tmp.name) / "chord.wav" + _write_chord_fixture(cls.fixture, sample_rate=22_050, duration_seconds=10.0) + + @classmethod + def tearDownClass(cls) -> None: + cls._tmp.cleanup() + + def test_transcribe_returns_non_empty_parseable_midi(self) -> None: + try: + import pretty_midi # noqa: F401 + except ImportError as exc: + self.skipTest(f"pretty_midi not installed: {exc}") + + try: + result = transcribe(self.fixture, stems_dir=None) + except Mt3NotAvailableError as exc: + self.skipTest( + f"MT3 extra not available: {exc}. Install via: " + "./apps/backend/venv/bin/pip install -r " + "apps/backend/requirements-mt3.txt" + ) + + self.assertIsInstance(result, Mt3Result) + self.assertTrue(result.version, "version must be a non-empty string") + self.assertGreater( + len(result.tracks), + 0, + "tracks must be non-empty for a sustained chord fixture", + ) + + import pretty_midi + for track in result.tracks: + midi_bytes = base64.b64decode(track.midi_b64) + parsed = pretty_midi.PrettyMIDI(BytesIO(midi_bytes)) + actual_note_count = sum( + len(instrument.notes) for instrument in parsed.instruments + ) + self.assertEqual( + actual_note_count, + track.note_count, + f"track {track.instrument!r}: declared noteCount " + f"{track.note_count} does not match parsed MIDI count " + f"{actual_note_count}", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/apps/backend/tests/test_phase2_citation_paths.py b/apps/backend/tests/test_phase2_citation_paths.py index 4a0fc31a..92b3a1a2 100644 --- a/apps/backend/tests/test_phase2_citation_paths.py +++ b/apps/backend/tests/test_phase2_citation_paths.py @@ -215,5 +215,88 @@ def test_invented_spectral_path_still_flagged_after_normalization(self): self.assertEqual(warnings[0]["code"], "UNRESOLVED_CITATION_PATH") +class Mt3CitationPathsTests(unittest.TestCase): + """F3: when an MT3 transcription result is supplied, the validator must + accept ``transcription.mt3.*`` citations (mirroring the frontend's + ``projectPhase1FromRun`` which folds the MT3 result into Phase1Result + under that namespace). When no MT3 result is supplied, the same citation + must be flagged — MT3 paths are only valid when MT3 actually ran. + """ + + MEASUREMENT = {"bpm": 128.0, "spectralBalance": {"subBass": 0.4}} + + # Stored MT3 shape (server.py::_execute_mt3_attempt). midiB64 is swapped + # for an artifact ref before persistence; there is no top-level noteCount. + MT3_RESULT = { + "version": "magenta-mt3-base", + "stemsUsed": ["bass", "other"], + "tracks": [ + { + "instrument": "bass", + "midiArtifactId": "artifact-123", + "midiSizeBytes": 412, + "noteCount": 96, + "pitchRange": [28, 52], + } + ], + } + + def test_mt3_citation_flagged_when_no_mt3_result_supplied(self): + phase2 = { + "abletonRecommendations": [ + {"device": "Operator", "phase1Fields": ["transcription.mt3.stemsUsed"]} + ] + } + warnings = _validate_phase2_citation_paths(phase2, self.MEASUREMENT) + self.assertEqual(len(warnings), 1) + self.assertEqual(warnings[0]["code"], "UNRESOLVED_CITATION_PATH") + self.assertEqual(warnings[0]["originalValue"], "transcription.mt3.stemsUsed") + + def test_mt3_top_level_citation_accepted_with_mt3_result(self): + phase2 = { + "abletonRecommendations": [ + {"device": "Operator", "phase1Fields": ["transcription.mt3.stemsUsed"]} + ] + } + warnings = _validate_phase2_citation_paths( + phase2, self.MEASUREMENT, mt3_result=self.MT3_RESULT + ) + self.assertEqual(warnings, []) + + def test_mt3_array_item_citation_accepted_with_mt3_result(self): + # Array-of-object fields register under `prefix.key` with no [i] index + # (same convention as arrangementDetail.noveltyPeaks.time). + phase2 = { + "mixAndMasterChain": [ + { + "device": "Operator", + "phase1Fields": [ + "transcription.mt3.tracks.instrument", + "transcription.mt3.tracks.noteCount", + ], + } + ] + } + warnings = _validate_phase2_citation_paths( + phase2, self.MEASUREMENT, mt3_result=self.MT3_RESULT + ) + self.assertEqual(warnings, []) + + def test_invented_mt3_path_still_flagged_with_mt3_result(self): + # Supplying mt3_result does not blanket-accept the namespace — a path + # absent from the actual MT3 payload is still flagged. + phase2 = { + "abletonRecommendations": [ + {"device": "Operator", "phase1Fields": ["transcription.mt3.bogusField"]} + ] + } + warnings = _validate_phase2_citation_paths( + phase2, self.MEASUREMENT, mt3_result=self.MT3_RESULT + ) + self.assertEqual(len(warnings), 1) + self.assertEqual(warnings[0]["code"], "UNRESOLVED_CITATION_PATH") + self.assertEqual(warnings[0]["originalValue"], "transcription.mt3.bogusField") + + if __name__ == "__main__": unittest.main() diff --git a/apps/backend/tests/test_server.py b/apps/backend/tests/test_server.py index 053d1cdb..f5a3a044 100644 --- a/apps/backend/tests/test_server.py +++ b/apps/backend/tests/test_server.py @@ -343,6 +343,50 @@ def test_build_phase2_prompt_includes_audio_observations_instructions(self) -> N self.assertIn("styleProfile", prompt) self.assertIn("authoritativeMeasurements", prompt) self.assertIn("category must be exactly one of", prompt) + + def test_build_phase2_prompt_renders_mt3_grounding_when_present(self) -> None: + # End-to-end proof of the F3 chain + the interpretation-ordering gate's + # payoff: when an MT3 result is supplied (which the gate now ensures has + # actually completed before interpretation runs), it must surface in the + # prompt under OPTIONAL_MT3_TRANSCRIPTION_RESULT_JSON so Gemini can cite + # transcription.mt3.* paths. The other prompt tests pass mt3_result=None. + mt3_result = { + "version": "magenta-mt3-base", + "stemsUsed": ["bass", "other"], + "tracks": [ + { + "instrument": "bass", + "midiArtifactId": "artifact-1", + "midiSizeBytes": 256, + "noteCount": 42, + "pitchRange": [28, 52], + } + ], + } + prompt = server._build_phase2_prompt( + measurement_result={"bpm": 128}, + pitch_note_result=None, + grounding_metadata={"profileId": "producer_summary"}, + descriptor_hooks=None, + mt3_result=mt3_result, + ) + + self.assertIn("OPTIONAL_MT3_TRANSCRIPTION_RESULT_JSON", prompt) + self.assertIn("magenta-mt3-base", prompt) + self.assertIn('"instrument": "bass"', prompt) + + def test_build_phase2_prompt_mt3_section_renders_null_when_absent(self) -> None: + # When no MT3 stage ran, the section is still declared (the prompt tells + # Gemini to ignore a null) rather than silently omitted — guards against + # a regression that drops the section and confuses the input contract. + prompt = server._build_phase2_prompt( + measurement_result={"bpm": 128}, + pitch_note_result=None, + grounding_metadata={"profileId": "producer_summary"}, + descriptor_hooks=None, + ) + + self.assertIn("OPTIONAL_MT3_TRANSCRIPTION_RESULT_JSON:\nnull", prompt) self.assertIn("workflowStage = the project phase", prompt) self.assertIn('"workflowStage":"SOUND_DESIGN","category":"SYNTHESIS"', prompt) self.assertIn("Return: with no space after the colon", prompt) @@ -1156,7 +1200,54 @@ def test_analysis_runs_estimate_endpoint_uses_staged_request_fields( [stage["key"] for stage in payload["estimate"]["stages"]], ["local_dsp", "demucs_separation", "transcription_stems"], ) - build_estimate_mock.assert_called_once_with(214.6, True, True, run_standard=True) + build_estimate_mock.assert_called_once_with( + 214.6, True, True, run_standard=True, run_mt3=False + ) + + @patch.object(server, "get_audio_duration_seconds", return_value=214.6, create=True) + @patch.object( + server, + "build_analysis_estimate", + return_value={ + "durationSeconds": 214.6, + "totalSeconds": {"min": 167, "max": 383}, + "stages": [ + { + "key": "dsp", + "label": "DSP analysis", + "seconds": {"min": 22, "max": 38}, + }, + { + "key": "mt3_transcription", + "label": "MT3 polyphonic transcription", + "seconds": {"min": 60, "max": 180}, + }, + ], + }, + create=True, + ) + def test_analysis_runs_estimate_endpoint_prices_mt3_when_enabled( + self, build_estimate_mock, *_mocks + ) -> None: + """F2: mt3_mode='enabled' must flow through to build_analysis_estimate as + run_mt3=True so the returned estimate includes the MT3 stage cost.""" + response = asyncio.run( + server.estimate_analysis_run( + track=self._upload_file(), + analysis_mode="standard", + pitch_note_mode="stem_notes", + pitch_note_backend="auto", + interpretation_mode="async", + interpretation_profile="producer_summary", + interpretation_model="gemini-2.5-flash", + mt3_mode="enabled", + ) + ) + + self.assertEqual(response.status_code, 200) + build_estimate_mock.assert_called_once_with( + 214.6, True, True, run_standard=True, run_mt3=True + ) def test_analysis_runs_estimate_endpoint_rejects_oversized_upload(self) -> None: with patch.object(server.upload_limits, "MAX_UPLOAD_SIZE_BYTES", 4): @@ -1228,6 +1319,48 @@ def test_analysis_runs_endpoint_rejects_unknown_pitch_note_backend(self) -> None self.assertEqual(payload["error"]["code"], "PITCH_NOTE_BACKEND_UNSUPPORTED") self.assertIn("mystery-backend", payload["error"]["message"]) + def test_analysis_runs_estimate_rejects_unknown_mt3_mode(self) -> None: + """An invalid mt3_mode at the estimate route returns a typed 400 rather + than silently pricing the run with MT3 off.""" + response = asyncio.run( + server.estimate_analysis_run( + track=self._upload_file(), + pitch_note_mode="off", + pitch_note_backend="auto", + interpretation_mode="off", + interpretation_profile="producer_summary", + interpretation_model=None, + mt3_mode="enable", + ) + ) + + self.assertEqual(response.status_code, 400) + payload = self._decode_json_response(response) + self.assertEqual(payload["error"]["code"], "MT3_MODE_UNSUPPORTED") + self.assertIn("enable", payload["error"]["message"]) + + def test_analysis_runs_endpoint_rejects_unknown_mt3_mode(self) -> None: + """An invalid mt3_mode at the create-run route returns a typed 400. This + path was previously dead: _coerce_mt3_mode silently coerced any unknown + value to 'off', so MT3_MODE_UNSUPPORTED could only fire via a direct + create_run() call, never over HTTP.""" + response = asyncio.run( + server.create_analysis_run( + track=self._upload_file(), + pitch_note_mode="off", + pitch_note_backend="auto", + interpretation_mode="off", + interpretation_profile="producer_summary", + interpretation_model=None, + mt3_mode="enable", + ) + ) + + self.assertEqual(response.status_code, 400) + payload = self._decode_json_response(response) + self.assertEqual(payload["error"]["code"], "MT3_MODE_UNSUPPORTED") + self.assertIn("enable", payload["error"]["message"]) + def test_analysis_runs_estimate_rejects_unknown_interpretation_profile(self) -> None: response = asyncio.run( server.estimate_analysis_run( @@ -1644,7 +1777,7 @@ def test_estimate_endpoint_combines_separate_and_transcribe_flags( [stage["key"] for stage in payload["estimate"]["stages"]], ["local_dsp", "demucs_separation", "transcription_stems"], ) - build_estimate_mock.assert_called_once_with(214.6, True, True) + build_estimate_mock.assert_called_once_with(214.6, True, True, run_mt3=False) @patch.object(server, "get_audio_duration_seconds", return_value=214.6, create=True) @patch.object( @@ -1754,7 +1887,7 @@ def test_analyze_endpoint_combines_separate_and_transcribe_in_subprocess( ], ) self.assertEqual(call_kwargs["timeout"], 526) - build_estimate_mock.assert_called_once_with(214.6, True, False) + build_estimate_mock.assert_called_once_with(214.6, True, False, run_mt3=False) payload = self._decode_json_response(response) self.assertEqual(payload["diagnostics"]["backendDurationMs"], 200.0) self.assertEqual( @@ -5297,5 +5430,355 @@ def test_get_audio_mime_type_prefers_canonical_audio_types(self) -> None: self.assertEqual(server._get_audio_mime_type("track.aiff"), "audio/aiff") +class Mt3ExecutorTests(unittest.TestCase): + """Unit tests for server._execute_mt3_attempt. + + Mocks the subprocess so the test does not require MT3 / JAX / t5x + installed. Each test exercises one branch of the executor: happy + path (artifact-ref swap), Mt3NotAvailableError classification, and + generic failure classification. These are the gap-closure tests + flagged by the post-MVP advisor review. + """ + + def _create_queued_mt3_run(self, runtime) -> tuple[str, str]: + """Helper: create a run, complete its measurement, and queue an + mt3 attempt. Returns (run_id, attempt_id).""" + created = runtime.create_run( + filename="track.wav", + content=b"fake-audio", + mime_type="audio/wav", + pitch_note_mode="off", + pitch_note_backend="auto", + interpretation_mode="off", + interpretation_profile="producer_summary", + interpretation_model=None, + mt3_mode="enabled", + ) + run_id = created["runId"] + # complete_measurement is what `_enqueue_requested_followups` hooks + # into to queue the mt3 attempt. + runtime.complete_measurement( + run_id, + payload={"bpm": 120.0}, + provenance={}, + diagnostics={}, + ) + # The followup is enqueued in `complete_measurement`. Look it up. + run_snapshot = runtime.get_run(run_id) + mt3_attempts = run_snapshot["stages"]["mt3"]["attemptsSummary"] + self.assertEqual(len(mt3_attempts), 1, "mt3 attempt should be queued") + attempt_id = mt3_attempts[0]["attemptId"] + # Reserve it so the executor's flow matches the worker loop. + runtime.reserve_mt3_attempt(attempt_id) + return run_id, attempt_id + + def test_happy_path_swaps_midi_b64_for_artifact_ref(self) -> None: + """The executor MUST decode midiB64 from the subprocess output, + persist it as an artifact, and replace the inline base64 with + midiArtifactId + midiSizeBytes in the stored result. Snapshot + polls must never carry the full MIDI bytes.""" + import base64 as _b64 + from analysis_runtime import AnalysisRuntime + + # Two synthetic MIDI bodies of different sizes so we can verify + # the size field is per-track, not a constant. + bass_midi = b"\x4d\x54\x68\x64" + b"\x00" * 12 # "MThd" + padding + other_midi = b"\x4d\x54\x68\x64" + b"\x01" * 40 + + mock_payload = { + "version": "mt3-py-0.1.0+magenta-mt3-base", + "stemsUsed": ["bass", "other"], + "tracks": [ + { + "instrument": "bass", + "midiB64": _b64.b64encode(bass_midi).decode("ascii"), + "noteCount": 12, + "pitchRange": [36, 60], + }, + { + "instrument": "other", + "midiB64": _b64.b64encode(other_midi).decode("ascii"), + "noteCount": 24, + "pitchRange": [48, 84], + }, + ], + } + + with tempfile.TemporaryDirectory(prefix="asa_mt3_executor_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + run_id, attempt_id = self._create_queued_mt3_run(runtime) + + def fake_subprocess_run(command, **kwargs): + return subprocess.CompletedProcess( + args=command, + returncode=0, + stdout=json.dumps(mock_payload), + stderr="", + ) + + with patch.object(server.subprocess, "run", side_effect=fake_subprocess_run): + server._execute_mt3_attempt( + runtime, + { + "attemptId": attempt_id, + "runId": run_id, + "checkpointId": "magenta-mt3-base", + }, + ) + + snapshot = runtime.get_run(run_id) + mt3_stage = snapshot["stages"]["mt3"] + self.assertEqual(mt3_stage["status"], "completed") + result = mt3_stage["result"] + self.assertIsNotNone(result) + self.assertEqual(result["version"], "mt3-py-0.1.0+magenta-mt3-base") + self.assertEqual(result["stemsUsed"], ["bass", "other"]) + self.assertEqual(len(result["tracks"]), 2) + + # Load-bearing assertion: no track carries midiB64; each + # carries midiArtifactId + midiSizeBytes. + for track in result["tracks"]: + self.assertNotIn( + "midiB64", + track, + f"Track {track['instrument']!r} still carries inline " + "midiB64 — snapshot polls would balloon. The executor " + "must swap to midiArtifactId + midiSizeBytes.", + ) + self.assertIn("midiArtifactId", track) + self.assertIn("midiSizeBytes", track) + self.assertIsNotNone(track["midiArtifactId"]) + + # Each track's midiSizeBytes matches the corresponding fixture + # length (verifies sizes are computed per-track, not a constant). + track_by_instrument = {t["instrument"]: t for t in result["tracks"]} + self.assertEqual(track_by_instrument["bass"]["midiSizeBytes"], len(bass_midi)) + self.assertEqual(track_by_instrument["other"]["midiSizeBytes"], len(other_midi)) + + # Two artifacts written, one per track, kinds match the + # mt3_track_ pattern. + mt3_artifacts = runtime.get_internal_artifacts_by_kind(run_id, "mt3_track_") + self.assertEqual(len(mt3_artifacts), 2) + artifact_kinds = sorted(a["kind"] for a in mt3_artifacts) + self.assertEqual(artifact_kinds, ["mt3_track_bass", "mt3_track_other"]) + + def test_mt3_not_available_marker_classifies_as_not_retryable(self) -> None: + """When the subprocess stderr matches the Mt3NotAvailableError + marker phrase, the attempt fails with MT3_NOT_AVAILABLE and + retryable=False — distinct from generic transcription failures.""" + from analysis_runtime import AnalysisRuntime + + with tempfile.TemporaryDirectory(prefix="asa_mt3_unavail_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + run_id, attempt_id = self._create_queued_mt3_run(runtime) + + def fake_subprocess_run(command, **kwargs): + return subprocess.CompletedProcess( + args=command, + returncode=1, + stdout="", + stderr=( + "Traceback (most recent call last):\n" + " File ...\n" + "mt3_transcription.Mt3NotAvailableError: " + "MT3 backend not installed. Install via: ...\n" + ), + ) + + with patch.object(server.subprocess, "run", side_effect=fake_subprocess_run): + server._execute_mt3_attempt( + runtime, + { + "attemptId": attempt_id, + "runId": run_id, + "checkpointId": "magenta-mt3-base", + }, + ) + + snapshot = runtime.get_run(run_id) + mt3_stage = snapshot["stages"]["mt3"] + self.assertEqual(mt3_stage["status"], "failed") + error = mt3_stage["error"] + self.assertIsNotNone(error) + self.assertEqual(error["code"], "MT3_NOT_AVAILABLE") + self.assertFalse(error["retryable"]) + + def test_generic_subprocess_failure_classifies_as_retryable(self) -> None: + """A non-zero exit whose stderr does NOT match the + Mt3NotAvailableError markers should classify as a generic + MT3_TRANSCRIPTION_FAILED with retryable=True so the operator + can re-enqueue.""" + from analysis_runtime import AnalysisRuntime + + with tempfile.TemporaryDirectory(prefix="asa_mt3_generic_fail_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + run_id, attempt_id = self._create_queued_mt3_run(runtime) + + def fake_subprocess_run(command, **kwargs): + return subprocess.CompletedProcess( + args=command, + returncode=139, # OOM / segfault-class exit + stdout="", + stderr="Process aborted: out of memory while loading audio\n", + ) + + with patch.object(server.subprocess, "run", side_effect=fake_subprocess_run): + server._execute_mt3_attempt( + runtime, + { + "attemptId": attempt_id, + "runId": run_id, + "checkpointId": "magenta-mt3-base", + }, + ) + + snapshot = runtime.get_run(run_id) + mt3_stage = snapshot["stages"]["mt3"] + self.assertEqual(mt3_stage["status"], "failed") + error = mt3_stage["error"] + self.assertIsNotNone(error) + self.assertEqual(error["code"], "MT3_TRANSCRIPTION_FAILED") + self.assertTrue(error["retryable"]) + + def test_mt3_timeout_expired_terminalizes_attempt(self) -> None: + """subprocess.run(timeout=1800) raises subprocess.TimeoutExpired, which + the executor's catch-all must terminalize to a 'failed' attempt rather + than leaving it 'running'. This is the linchpin of the no-deadlock claim + for the interpretation-ordering gate: a hung MT3 must not block a queued + interpretation forever — a 'failed' (terminal) mt3 unblocks it.""" + from analysis_runtime import AnalysisRuntime + + with tempfile.TemporaryDirectory(prefix="asa_mt3_timeout_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + run_id, attempt_id = self._create_queued_mt3_run(runtime) + + with patch.object( + server.subprocess, "run", side_effect=_make_timeout_expired() + ): + server._execute_mt3_attempt( + runtime, + { + "attemptId": attempt_id, + "runId": run_id, + "checkpointId": "magenta-mt3-base", + }, + ) + + snapshot = runtime.get_run(run_id) + mt3_stage = snapshot["stages"]["mt3"] + self.assertEqual(mt3_stage["status"], "failed") + error = mt3_stage["error"] + self.assertIsNotNone(error) + self.assertEqual(error["code"], "MT3_TRANSCRIPTION_FAILED") + + def test_source_artifact_failure_terminalizes_attempt(self) -> None: + """If get_source_artifact raises (e.g. a missing artifact row), the + executor must terminalize the attempt to 'failed' rather than leave it + stuck 'running'. get_source_artifact runs inside the try precisely so + this can't escape: with the interpretation-ordering gate, a stuck-'running' + mt3 would block this run's interpretation indefinitely. The except + diagnostics are null-safe (source_artifact is None on this path).""" + from analysis_runtime import AnalysisRuntime + + with tempfile.TemporaryDirectory(prefix="asa_mt3_src_fail_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + run_id, attempt_id = self._create_queued_mt3_run(runtime) + + with patch.object( + runtime, "get_source_artifact", side_effect=KeyError("missing source artifact") + ): + server._execute_mt3_attempt( + runtime, + { + "attemptId": attempt_id, + "runId": run_id, + "checkpointId": "magenta-mt3-base", + }, + ) + + snapshot = runtime.get_run(run_id) + mt3_stage = snapshot["stages"]["mt3"] + # The key assertion: terminal, NOT stuck 'running'. + self.assertEqual(mt3_stage["status"], "failed") + self.assertEqual(mt3_stage["error"]["code"], "MT3_TRANSCRIPTION_FAILED") + + def test_mt3_stage_public_status_present_when_not_requested(self) -> None: + """Sanity check for the snapshot contract: stages.mt3.publicStatus + is present on every snapshot (not just when MT3 ran). When + mt3_mode is the default 'off', status is 'not_requested' and + publicStatus is None — the additive 5-value contract. + + This is the assertion the post-MVP advisor flagged as unverified. + """ + from analysis_runtime import AnalysisRuntime + + with tempfile.TemporaryDirectory(prefix="asa_mt3_pubstatus_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + created = runtime.create_run( + filename="track.wav", + content=b"fake-audio", + mime_type="audio/wav", + pitch_note_mode="off", + pitch_note_backend="auto", + interpretation_mode="off", + interpretation_profile="producer_summary", + interpretation_model=None, + # mt3_mode defaults to "off" + ) + snapshot = runtime.get_run(created["runId"]) + self.assertIn("mt3", snapshot["stages"]) + mt3_stage = snapshot["stages"]["mt3"] + self.assertEqual(mt3_stage["status"], "not_requested") + self.assertFalse(mt3_stage["authoritative"]) + self.assertEqual(mt3_stage["attemptsSummary"], []) + self.assertIsNone(mt3_stage["result"]) + # publicStatus is annotated on the HTTP layer by server_phase1.py; + # the raw runtime snapshot doesn't have it. The HTTP route's + # _normalize_run_snapshot wraps in _annotate_public_status and + # adds it. Confirm presence of the underlying status here; + # publicStatus contract is exercised by the next test. + + def test_mt3_stage_public_status_annotated_after_normalization(self) -> None: + """_annotate_public_status iterates stages.items() and adds + publicStatus to every dict-typed stage entry. Verify that adding + a new stage to the snapshot is automatically picked up.""" + from analysis_runtime import AnalysisRuntime + from server_phase1 import _annotate_public_status, _normalize_run_snapshot + + with tempfile.TemporaryDirectory(prefix="asa_mt3_annot_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + created = runtime.create_run( + filename="track.wav", + content=b"fake-audio", + mime_type="audio/wav", + pitch_note_mode="off", + pitch_note_backend="auto", + interpretation_mode="off", + interpretation_profile="producer_summary", + interpretation_model=None, + mt3_mode="enabled", + ) + run_snapshot = runtime.get_run(created["runId"]) + normalized = _normalize_run_snapshot(run_snapshot, runtime) + self.assertIn("mt3", normalized["stages"]) + mt3_stage = normalized["stages"]["mt3"] + self.assertIn( + "publicStatus", + mt3_stage, + "_annotate_public_status must add publicStatus to the mt3 stage. " + "If this test fails, the public-status annotator no longer " + "iterates stages dynamically — re-check server_phase1.py.", + ) + # mt3_mode='enabled' creates a queued attempt automatically when + # measurement completes; until then, the stage status reflects + # "blocked" (measurement not yet completed). publicStatus + # collapses to "queued" per the 5-value contract. + self.assertIn( + mt3_stage["publicStatus"], + {"queued", "running", "completed", "failed", "interrupted", None}, + f"publicStatus must be one of the 5-value enum or None; got {mt3_stage['publicStatus']!r}", + ) + + if __name__ == "__main__": unittest.main() diff --git a/apps/ui/src/App.tsx b/apps/ui/src/App.tsx index bad8235a..a4953ea1 100644 --- a/apps/ui/src/App.tsx +++ b/apps/ui/src/App.tsx @@ -15,6 +15,7 @@ import { useGlobalDrag } from './hooks/useGlobalDrag'; import { appConfig, isGeminiPhase2ConfigEnabled, + isMt3ConfigEnabled, } from './config'; import { getAudioMimeTypeOrDefault, isSupportedAudioFile } from './services/audioFile'; import { analyzeAudio, monitorAnalysisRun } from './services/analyzer'; @@ -265,12 +266,14 @@ export default function App() { const [elapsedMs, setElapsedMs] = useState(0); const [analysisMode, setAnalysisMode] = useState<'full' | 'standard'>('full'); const [pitchNoteTranslationRequested, setPitchNoteTranslationRequested] = useState(true); + const [mt3Requested, setMt3Requested] = useState(false); const analysisStartedAtRef = useRef(null); const abortControllerRef = useRef(null); const currentRunIdRef = useRef(null); const ignoredRunIdsRef = useRef>(new Set()); const phase2ConfigEnabled = isGeminiPhase2ConfigEnabled(); + const mt3ConfigEnabled = isMt3ConfigEnabled(); const interpretationWillRun = interpretationRequested && phase2ConfigEnabled; const phase2StatusBadge = getInterpretationStatusBadge(phase2ConfigEnabled, interpretationRequested); const phase2HelperCopy = getInterpretationHelperCopy(phase2ConfigEnabled, interpretationRequested); @@ -333,6 +336,7 @@ export default function App() { interpretationMode: interpretationWillRun ? 'async' : 'off', interpretationProfile: 'producer_summary', interpretationModel: interpretationWillRun ? selectedModel : undefined, + mt3Mode: mt3Requested && mt3ConfigEnabled ? 'enabled' : 'off', }) .then((result) => { if (isCancelled) return; @@ -353,7 +357,7 @@ export default function App() { return () => { isCancelled = true; }; - }, [analysisMode, audioFile, interpretationWillRun, pitchNoteTranslationRequested, selectedModel]); + }, [analysisMode, audioFile, interpretationWillRun, mt3Requested, mt3ConfigEnabled, pitchNoteTranslationRequested, selectedModel]); useEffect(() => { if (!isAnalyzing || analysisStartedAtRef.current === null) { @@ -664,6 +668,7 @@ export default function App() { { analysisMode, pitchNoteRequested: activePitchNoteRequested, + mt3Requested, timeoutMs: activeTimeoutMs, signal: ac.signal, interpretationRequested, @@ -1182,6 +1187,36 @@ export default function App() { + {mt3ConfigEnabled && ( + + )}