refactor(engine): promote temporal probe from cli hack to api integration

apps/desktop/core/src/lib.rs

@@ -34,7 +34,7 @@ pub struct AppStateInner {
 
 pub const MAX_IN_FLIGHT_JOBS: usize = 2;
 
-pub const ANALYSIS_PROCESS_TIMEOUT: Duration = Duration::from_secs(30);
+pub const ANALYSIS_PROCESS_TIMEOUT: Duration = Duration::from_secs(360);
 
 pub const ANALYSIS_WAIT_POLL: Duration = Duration::from_millis(50);
 

services/analysis-engine/src/bandscope_analysis/api.py

@@ -19,13 +19,14 @@
 from bandscope_analysis.sections import extract_sections
 from bandscope_analysis.sections.segmenter import segment_with_boundaries
 from bandscope_analysis.separation import AudioStemSeparator
+from bandscope_analysis.temporal import TemporalAnalyzer
 
 logger = logging.getLogger(__name__)
 
 MAX_SECTION_TIME_SECONDS = 4_294_967_295
-ANALYSIS_CACHE_SCHEMA_VERSION = 1
+ANALYSIS_CACHE_SCHEMA_VERSION = 2
 FEATURE_CACHE_SCHEMA_VERSION = 1
-STEM_SEPARATION_TIMEOUT_SECONDS = 20.0
+STEM_SEPARATION_TIMEOUT_SECONDS = 300.0
 
 logger = logging.getLogger(__name__)
 
@@ -472,7 +473,9 @@ def _build_from_arrangement(audio_features: dict[str, Any] | None = None) -> Reh
 
     song: RehearsalSong = {
         "id": "demo-song",
-        "title": "Late Night Set",
+        "title": (
+            audio_features.get("title", "Late Night Set") if audio_features else "Late Night Set"
+        ),
         "sections": [
             {
                 "id": verse_section["id"],
@@ -613,15 +616,29 @@ def _analysis_cache_path(request: AnalysisJobRequest) -> Path | None:
     digest = hashlib.sha256(
         json.dumps(key_payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
     ).hexdigest()
-    return Path(cache_root) / "analysis-cache-v1" / f"{digest}.json"
+    return Path(cache_root) / "analysis-cache-v2" / f"{digest}.json"
 
 
 def _feature_cache_paths(request: AnalysisJobRequest) -> tuple[Path, Path] | None:
     """Return metadata + array cache paths for intermediate local-audio features."""
-    analysis_cache_path = _analysis_cache_path(request)
-    if analysis_cache_path is None:
+    if request["sourceKind"] != "local_audio" or "localSource" not in request:
+        return None
+    cache_root = request.get("cacheRoot")
+    if not cache_root:
         return None
-    stem_cache_base = analysis_cache_path.with_suffix("")
+
+    local_source = request["localSource"]
+    key_payload = {
+        "schemaVersion": FEATURE_CACHE_SCHEMA_VERSION,
+        "projectId": request.get("projectId", ""),
+        "sourcePath": local_source["sourcePath"],
+        "fileName": local_source["fileName"],
+        "fileSizeBytes": local_source["fileSizeBytes"],
+    }
+    digest = hashlib.sha256(
+        json.dumps(key_payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    ).hexdigest()
+    stem_cache_base = Path(cache_root) / "analysis-cache-v1" / digest
     return (
         stem_cache_base.with_suffix(".features.json"),
         stem_cache_base.with_suffix(".features.npz"),
@@ -1039,6 +1056,21 @@ def _build_local_audio_features(request: AnalysisJobRequest) -> dict[str, Any] |
     }
 
 
+def _build_local_temporal_features(request: AnalysisJobRequest) -> dict[str, Any] | None:
+    """Extract tempo features that remain useful when stem separation is unavailable."""
+    if request["sourceKind"] != "local_audio" or "localSource" not in request:
+        return None
+
+    try:
+        return {
+            "title": request["sourceLabel"],
+            **dict(TemporalAnalyzer().analyze(request["localSource"]["sourcePath"])),
+        }
+    except (FileNotFoundError, ValueError):
+        logger.warning("Temporal analysis unavailable; continuing with safe fallback.")
+        return {"title": request["sourceLabel"]}
+
+
 def run_analysis_job_updates(
     job_id: str,
     payload: object,
@@ -1102,12 +1134,13 @@ def run_analysis_job_updates(
             cache_status=cache_status,
         ),
     ]
-    audio_features: dict[str, Any] | None = None
+    temporal_features: dict[str, Any] | None = None
+    stem_features: dict[str, Any] | None = None
     feature_cache_hit = False
     if feature_cache_paths is not None:
         cached_features = _load_cached_local_audio_features(*feature_cache_paths)
         if cached_features is not None:
-            audio_features = cached_features
+            stem_features = cached_features
             feature_cache_hit = True
             updates.append(
                 _build_job_status(
@@ -1121,7 +1154,7 @@ def run_analysis_job_updates(
                 )
             )
 
-    if audio_features is None:
+    if stem_features is None:
         updates.append(
             _build_job_status(
                 job_id=job_id,
@@ -1134,7 +1167,7 @@ def run_analysis_job_updates(
             )
         )
         try:
-            audio_features = _build_local_audio_features(request)
+            stem_features = _build_local_audio_features(request)
         except StemSeparationTimedOut:
             updates.append(
                 _build_job_status(
@@ -1147,7 +1180,7 @@ def run_analysis_job_updates(
                     cache_status=cache_status,
                 )
             )
-            audio_features = None
+            stem_features = None
         except RuntimeError:
             updates.append(
                 _build_job_status(
@@ -1160,7 +1193,7 @@ def run_analysis_job_updates(
                     cache_status=cache_status,
                 )
             )
-            audio_features = None
+            stem_features = None
         except (FileNotFoundError, ValueError):
             logger.exception("Stem separation failed before analysis job completion.")
             updates.append(
@@ -1180,6 +1213,14 @@ def run_analysis_job_updates(
             )
             return updates
 
+    temporal_features = _build_local_temporal_features(request)
+    audio_features: dict[str, Any] | None = None
+    if temporal_features is not None or stem_features is not None:
+        audio_features = {
+            **(temporal_features or {}),
+            **(stem_features or {}),
+        }
+
     updates.append(
         _build_job_status(
             job_id=job_id,

services/analysis-engine/src/bandscope_analysis/cli.py

@@ -8,7 +8,6 @@
 from datetime import UTC, datetime
 
 from bandscope_analysis.api import get_analysis_status, run_analysis_job, run_analysis_job_updates
-from bandscope_analysis.temporal import TemporalAnalyzer
 
 logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
 
@@ -75,28 +74,6 @@ def main() -> int:
 
     request = payload.get("request")
 
-    # Temporary: Inject temporal analyzer call if it's a local file, just to prove it works
-    # before full orchestrator integration
-    if (
-        isinstance(request, dict)
-        and request.get("sourceKind") == "local_audio"
-        and "localSource" in request
-    ):
-        local_source = request["localSource"]
-        audio_path = local_source.get("sourcePath")
-        file_name = local_source.get("fileName", "selected audio")
-        if audio_path:
-            logging.info("Extracting temporal features from %s...", file_name)
-            try:
-                temporal_analyzer = TemporalAnalyzer()
-                features = temporal_analyzer.analyze(audio_path)
-                logging.info(f"Extracted BPM: {features['bpm']}")
-            except Exception:
-                logging.warning(
-                    "Temporal analysis failed for %s; continuing with safe fallback.",
-                    file_name,
-                )
-
     requested_at = datetime.now(UTC).isoformat().replace("+00:00", "Z")
     if progress_jsonl:
         for update in run_analysis_job_updates(job_id, request, requested_at):

services/analysis-engine/tests/test_analysis_cache_v2_malformed_result.py

@@ -0,0 +1,11 @@
+"""Regression coverage for malformed current-schema analysis cache entries."""
+
+from bandscope_analysis.api import _load_cached_analysis
+
+
+def test_current_schema_cache_rejects_non_mapping_result(tmp_path) -> None:
+    """Treat an untrusted v2 cache payload with a non-object result as a cache miss."""
+    cache_path = tmp_path / "analysis-cache-v2.json"
+    cache_path.write_text('{"schemaVersion": 2, "result": []}', encoding="utf-8")
+
+    assert _load_cached_analysis(cache_path) is None

services/analysis-engine/tests/test_api.py

@@ -396,8 +396,9 @@ def test_build_demo_rehearsal_song_matches_expected_fixture() -> None:
 
 def test_build_demo_rehearsal_song_with_tempo() -> None:
     """Ensure build_demo_rehearsal_song incorporates tempo from audio features."""
-    song = build_demo_rehearsal_song({"bpm": 120.4})
+    song = build_demo_rehearsal_song({"bpm": 120.4, "title": "actual-recording.wav"})
     assert song.get("tempo") == 120
+    assert song["title"] == "actual-recording.wav"
 
 
 def test_coerce_tempo_bpm() -> None:
@@ -582,9 +583,12 @@ def test_run_analysis_job_updates_report_progress_and_cache(tmp_path) -> None:
             ("succeeded", "ready", 100),
         ]
         assert updates[-1]["cacheStatus"] == "stored"
-        cache_files = list((tmp_path / "cache" / "analysis-cache-v1").glob("*.json"))
-        assert len([path for path in cache_files if not path.name.endswith(".features.json")]) == 1
-        assert len([path for path in cache_files if path.name.endswith(".features.json")]) == 1
+        result_cache_files = list((tmp_path / "cache" / "analysis-cache-v2").glob("*.json"))
+        feature_cache_files = list(
+            (tmp_path / "cache" / "analysis-cache-v1").glob("*.features.json")
+        )
+        assert len(result_cache_files) == 1
+        assert len(feature_cache_files) == 1
 
         cached_updates = list(
             run_analysis_job_updates("job-cache-2", payload, "2026-03-12T00:00:00Z")
@@ -1333,10 +1337,14 @@ def join(self, timeout: float | None = None) -> None:
 
 def test_run_analysis_job_updates_degrades_when_stem_step_is_unavailable() -> None:
     """Ensure runtime ML failures continue with fallback cues."""
-    with patch(
-        "bandscope_analysis.api._build_local_audio_features",
-        side_effect=RuntimeError("oom"),
+    with (
+        patch(
+            "bandscope_analysis.api._build_local_audio_features",
+            side_effect=RuntimeError("oom"),
+        ),
+        patch("bandscope_analysis.api.TemporalAnalyzer") as temporal_analyzer,
     ):
+        temporal_analyzer.return_value.analyze.return_value = {"bpm": 156.605}
         updates = list(
             run_analysis_job_updates(
                 "job-runtime",
@@ -1357,6 +1365,7 @@ def test_run_analysis_job_updates_degrades_when_stem_step_is_unavailable() -> No
         )
 
     assert updates[-1]["state"] == "succeeded"
+    assert updates[-1]["result"]["tempo"] == 157
     assert any(
         update.get("progressLabel") == "Stem separation unavailable; continuing with fallback cues"
         for update in updates

services/analysis-engine/tests/test_branch_coverage_contract.py

@@ -2,16 +2,12 @@
 
 from __future__ import annotations
 
-import io
-import json
 from pathlib import Path
 from unittest.mock import patch
 
 import numpy as np
-import pytest
 
 from bandscope_analysis import api as analysis_api
-from bandscope_analysis import cli
 from bandscope_analysis.chords.analyzer import ChordAnalyzer
 from bandscope_analysis.chords.chord_recognizer import ChordRecognizer
 from bandscope_analysis.exports import chart
@@ -95,34 +91,35 @@ def test_chord_segment_builder_handles_zero_frames_without_final_segment() -> No
     assert result == []
 
 
-def test_cli_skips_temporal_probe_when_local_source_path_is_empty(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    """Do not invoke the temporary temporal probe for an empty local source path."""
-    payload = {
-        "jobId": "job-empty-source",
-        "request": {
-            "sourceKind": "local_audio",
-            "localSource": {"sourcePath": "", "fileName": "song.wav"},
-        },
-    }
-    stdout = io.StringIO()
-    monkeypatch.setattr(cli.sys, "argv", ["cli.py"])
-    monkeypatch.setattr(cli.sys, "stdin", io.StringIO(json.dumps(payload)))
-    monkeypatch.setattr(cli.sys, "stdout", stdout)
+def test_temporal_feature_builder_returns_none_for_non_local_or_missing_source() -> None:
+    """Skip temporal extraction unless the request carries a local-audio source."""
+    assert (
+        analysis_api._build_local_temporal_features(
+            {"sourceKind": "youtube", "sourceLabel": "clip"}
+        )
+        is None
+    )
+    assert (
+        analysis_api._build_local_temporal_features(
+            {"sourceKind": "local_audio", "sourceLabel": "clip"}
+        )
+        is None
+    )
 
-    with (
-        patch.object(cli, "TemporalAnalyzer") as temporal_analyzer,
-        patch.object(
-            cli,
-            "run_analysis_job",
-            return_value={"jobId": "job-empty-source", "state": "failed"},
-        ),
-    ):
-        assert cli.main() == 0
 
-    temporal_analyzer.assert_not_called()
-    assert json.loads(stdout.getvalue())["jobId"] == "job-empty-source"
+def test_temporal_feature_builder_recovers_when_source_path_cannot_be_read() -> None:
+    """Keep the source label when the analyzer cannot open the recording path."""
+    request = {
+        "sourceKind": "local_audio",
+        "sourceLabel": "song.wav",
+        "localSource": {"sourcePath": "", "fileName": "song.wav"},
+    }
+    with patch.object(
+        analysis_api,
+        "TemporalAnalyzer",
+        side_effect=FileNotFoundError("no such file"),
+    ):
+        assert analysis_api._build_local_temporal_features(request) == {"title": "song.wav"}
 
 
 def test_chart_section_without_active_roles_and_duplicate_priority_footer() -> None: