feat: analysis jobs queue inference xai ve timeline modules ile unit tests eklendi

app/routes/analyze.py

@@ -50,6 +50,7 @@ from .analyze_schemas import (
     MetaClassifierExplanation,
     MetaFeatureImportance,
     ModelVoteModel,
+    Timeline,
     VocalAnalysis,
     XAIExplanation,
 )
@@ -70,10 +71,13 @@ from ..services.analysis_jobs import (
     result_cache,
 )
 from ..services.feature_extractor import (
+    MAX_TRACK_SEC,
     decode_clip,
     extract_features,
+    load_track,
     AudioFeatures,
 )
+from ..services.timeline import scan_track
 from ..services.vocal_analyzer import (
     analyze_vocals,
     VocalFeatures,
@@ -101,12 +105,12 @@ _EXT_MIME = {
 }
 URL_SOURCES = ("youtube", "tiktok", "instagram", "soundcloud", "twitter")
 
-# A link only needs its opening: features read 60 s, wav2vec2 30 s.
-YOUTUBE_CLIP_SEC = 120.0
+# A link needs the part the timeline scan reads; the clip layers use its first 60 s.
+YOUTUBE_CLIP_SEC = MAX_TRACK_SEC
 YOUTUBE_MAX_BYTES = 100 * 1024 * 1024
 
 # Bump when the pipeline's output changes, so results cached from the old one aren't served.
-PIPELINE_VERSION = "2"
+PIPELINE_VERSION = "3"
 
 # Rate limiter for public heavy endpoints
 _analyze_rate_store: dict[str, list[float]] = defaultdict(list)
@@ -186,6 +190,39 @@ def _probe_audio(content: bytes) -> Tuple[Optional[int], Optional[int], Optional
         return None, None, None
 
 
+async def _scan_timeline(
+    request_id: str,
+    content: bytes,
+    total_duration: Optional[float],
+    progress: Progress,
+    loop: "asyncio.AbstractEventLoop",
+) -> Optional[Dict[str, Any]]:
+    """Read the whole track window by window. Never fails the analysis: on any
+    problem the result simply has no timeline."""
+    progress.set("timeline", STEP_RUNNING)
+    try:
+        y, sr = await loop.run_in_executor(_analysis_pool, load_track, content)
+        total = total_duration or _probe_audio(content)[2] or y.size / sr
+        payload = await loop.run_in_executor(
+            _analysis_pool,
+            lambda: scan_track(
+                y, sr, xai_service,
+                total_sec=total,
+                on_progress=lambda done, n: progress.advance("timeline", done, n),
+                cancelled=lambda: progress.cancelled,
+            ),
+        )
+        progress.set("timeline", STEP_DONE)
+        return payload
+    except ValueError as e:
+        logger.info(f"[{request_id}] Timeline skipped: {e}")
+        progress.set("timeline", STEP_SKIPPED)
+    except Exception as e:  # noqa: BLE001 - the verdict stands without it
+        logger.warning(f"[{request_id}] Timeline failed: {e}", exc_info=True)
+        progress.set("timeline", STEP_FAILED)
+    return None
+
+
 def _track(future: "asyncio.Future[Any]", progress: Progress, step: str,
            ok: Optional[Callable[[Any], bool]] = None) -> None:
     """Report a parallel step as done/skipped/failed the moment it settles."""
@@ -243,6 +280,7 @@ async def _run_pipeline(
     logger.info(f"[{request_id}] Audio: {filename} ({file_size / 1024 / 1024:.2f}MB)")
     loop = asyncio.get_running_loop()
     fst_task: Optional["asyncio.Task[Any]"] = None
+    timeline_payload: Optional[Dict[str, Any]] = None
 
     try:
         async with cpu_gate.slot(ticket, progress):
@@ -276,6 +314,14 @@ async def _run_pipeline(
             _track(wav2vec2_future, progress, "wav2vec2", ok=lambda r: bool(r.available))
             # Hold the slot until every thread is done, even if one fails.
             await asyncio.gather(feat_future, vocal_future, clap_future, wav2vec2_future, return_exceptions=True)
+            progress.check()
+
+            # The rest of the track, window by window (same slot, same models).
+            scan = xai_service.available and feat_future.exception() is None
+            timeline_payload = await _scan_timeline(request_id, content, total_duration, progress, loop) if scan else None
+            if scan and timeline_payload is None:
+                warnings.append("timeline_unavailable")
+            progress.check()
         progress.check()
     except BaseException:
         if fst_task is not None:
@@ -549,6 +595,7 @@ async def _run_pipeline(
         xai=xai_payload,
         towerScores=tower_scores,
         metaClassifier=meta_payload,
+        timeline=Timeline(**timeline_payload) if timeline_payload else None,
         layers=layers,
     )
 

app/services/analysis_jobs.py

@@ -43,7 +43,7 @@ STEP_DONE = "done"
 STEP_FAILED = "failed"
 STEP_SKIPPED = "skipped"
 
-FILE_STEPS = ["features", "vocals", "wav2vec2", "clap", "fst", "xai", "meta"]
+FILE_STEPS = ["features", "vocals", "wav2vec2", "clap", "timeline", "fst", "xai", "meta"]
 URL_STEPS = ["download", *FILE_STEPS]
 
 
@@ -62,6 +62,8 @@ class Progress:
         self._steps: Dict[str, str] = {s: STEP_PENDING for s in steps}
         self._started: Dict[str, float] = {}
         self._took: Dict[str, float] = {}
+        # Steps made of many parts (the timeline scan) report how far along they are.
+        self._fraction: Dict[str, float] = {}
         self.cancelled = False
         # queued: waiting for the CPU; running: downloading or analysing.
         self.phase = "queued"
@@ -78,6 +80,10 @@ class Progress:
             self._took[step] = round(now - self._started[step], 2)
         self._steps[step] = state
 
+    def advance(self, step: str, done: int, total: int) -> None:
+        if step in self._steps and total > 0:
+            self._fraction[step] = round(min(1.0, done / total), 3)
+
     def running(self, *steps: str) -> None:
         for s in steps:
             self.set(s, STEP_RUNNING)
@@ -97,6 +103,8 @@ class Progress:
             item: Dict[str, Any] = {"id": step, "state": state}
             if step in self._took:
                 item["seconds"] = self._took[step]
+            if step in self._fraction and state == STEP_RUNNING:
+                item["progress"] = self._fraction[step]
             out.append(item)
         return out
 

app/services/inference_xai.py

@@ -408,6 +408,20 @@ class XAIResult:
     feature_count: int = 49
 
 
+@dataclass
+class WindowReading:
+    """One timeline window read by the primary classifier."""
+    probability: float
+    is_ai: bool
+    margin: float                       # 0 at the threshold, 1 at the far end of its side
+    reasons: List[FeatureContribution] = field(default_factory=list)
+
+
+# The classifier was trained on features of 60 s clips; a window's beat count is
+# scaled to that length so it lands on the same axis as the training data.
+_TRAINING_CLIP_SEC = 60.0
+
+
 class XAIInferenceService:
     """Loads trained artifacts and performs explainable inference."""
 
@@ -588,6 +602,70 @@ class XAIInferenceService:
             feature_count=len(self.feature_cols),
         )
 
+    def predict_window(
+        self,
+        features: AudioFeatures,
+        vocals: Optional[VocalFeatures],
+        *,
+        window_sec: float,
+        top: int = 3,
+    ) -> Optional[WindowReading]:
+        """Probability for one window of a longer track, with its main reasons.
+
+        Same LightGBM model, scaler and threshold as predict(); no ensemble vote
+        and no full SHAP table, since a timeline holds many windows.
+        """
+        if not self.available:
+            return None
+
+        feature_map = self._build_feature_map(features, vocals)
+        feature_map["beat_count"] = feature_map["beat_count"] * (_TRAINING_CLIP_SEC / max(window_sec, 1.0))
+        x = np.array([feature_map.get(col, 0.0) for col in self.feature_cols], dtype=np.float64)
+        x = np.nan_to_num(x, nan=0.0, posinf=1.0, neginf=-1.0)
+        x_scaled = self.scaler.transform(x.reshape(1, -1))
+
+        prob = float(self.model.predict_proba(x_scaled)[0, 1])
+        band = self._confidence_band(prob)
+
+        reasons: List[FeatureContribution] = []
+        if self.shap_explainer is not None:
+            try:
+                shap_values = self.shap_explainer.shap_values(x_scaled)
+                if isinstance(shap_values, list):
+                    sv = shap_values[1][0] if len(shap_values) > 1 else shap_values[0][0]
+                else:
+                    sv = shap_values[0]
+                ranked = sorted(range(len(self.feature_cols)), key=lambda i: abs(float(sv[i])), reverse=True)
+                for i in ranked[:top]:
+                    col = self.feature_cols[i]
+                    shap_v = float(sv[i])
+                    if abs(shap_v) < 0.001:
+                        continue
+                    raw = feature_map.get(col, 0.0)
+                    stats = self.feature_stats.get(col, {})
+                    std = stats.get("std", 1.0) or 1.0
+                    meta = FEATURE_CATALOG.get(col, {})
+                    reasons.append(FeatureContribution(
+                        name=col,
+                        label=meta.get("label", col),
+                        label_en=meta.get("labelEn", col),
+                        category=meta.get("category", "other"),
+                        value=float(raw),
+                        z_score=float((raw - stats.get("mean", 0.0)) / std),
+                        shap_value=shap_v,
+                        direction="towards_ai" if shap_v > 0 else "towards_human",
+                        description=meta.get("description", ""),
+                    ))
+            except Exception as e:  # noqa: BLE001 - a missing explanation must not drop the window
+                logger.warning(f"Window SHAP failed: {e}")
+
+        return WindowReading(
+            probability=prob,
+            is_ai=prob >= self.threshold,
+            margin=band.margin,
+            reasons=reasons,
+        )
+
     def _build_feature_map(
         self,
         features: AudioFeatures,

app/services/timeline.py

@@ -0,0 +1,180 @@
+"""Whole-track scan.
+
+The opening clip gives AURIS its verdict. This module reads the rest of the
+track: the first six minutes are cut into windows of about 30 s, every window
+goes through the same features and the same LightGBM model, and the readings
+come back as a timeline the page draws under the waveform.
+
+A window score is the classifier applied to an excerpt shorter than the 60 s
+clips it was trained on, and every training clip was either all AI or all
+human. A timeline therefore shows where the music reads AI-like, not where
+an AI took over.
+"""
+
+from __future__ import annotations
+
+import os
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor
+from typing import Any, Callable, Dict, List, Optional, Tuple
+
+import numpy as np
+
+from .feature_extractor import extract_features_array
+from .inference_xai import XAIInferenceService
+from .logging_config import get_logger
+from .vocal_analyzer import analyze_vocals_array
+
+logger = get_logger(__name__)
+
+TARGET_WINDOW_SEC = 30.0
+# A window quieter than this never reaches the classifier.
+SILENCE_PEAK = 1e-3
+PEAK_POINTS = 600
+# Windows not started within this many seconds are reported as skipped.
+SCAN_BUDGET_SEC = float(os.getenv("AURIS_SCAN_BUDGET_SEC", "180"))
+SCAN_WORKERS = max(1, int(os.getenv("AURIS_SCAN_WORKERS", "2")))
+
+
+def plan_windows(duration: float) -> List[Tuple[float, float]]:
+    """Equal windows of about 30 s that tile the whole duration (20 to 45 s each)."""
+    if duration <= 0:
+        return []
+    count = max(1, int(round(duration / TARGET_WINDOW_SEC)))
+    size = duration / count
+    return [
+        (round(i * size, 3), round(duration if i == count - 1 else (i + 1) * size, 3))
+        for i in range(count)
+    ]
+
+
+def waveform_peaks(y: np.ndarray, points: int = PEAK_POINTS) -> List[float]:
+    """Peak level per slice of the track, 0 to 1, for drawing the waveform."""
+    if y.size == 0:
+        return []
+    edges = np.linspace(0, y.size, min(points, y.size) + 1).astype(int)
+    peaks = np.array([
+        float(np.max(np.abs(y[a:b]))) if b > a else 0.0
+        for a, b in zip(edges[:-1], edges[1:])
+    ])
+    top = float(peaks.max())
+    if top <= 0:
+        return [0.0] * len(peaks)
+    return [round(float(v), 3) for v in peaks / top]
+
+
+def _read_window(
+    y: np.ndarray,
+    sr: int,
+    index: int,
+    start: float,
+    end: float,
+    xai: XAIInferenceService,
+) -> Dict[str, Any]:
+    base: Dict[str, Any] = {"index": index, "start": round(start, 2), "end": round(end, 2)}
+    seg = np.ascontiguousarray(y[int(start * sr):min(y.size, int(end * sr))])
+    if seg.size < sr:
+        return {**base, "state": "skipped"}
+    if float(np.max(np.abs(seg))) < SILENCE_PEAK:
+        return {**base, "state": "silent"}
+    try:
+        features = extract_features_array(seg, sr)
+        vocals = None
+        try:
+            vocals = analyze_vocals_array(seg, sr)
+        except Exception as e:  # noqa: BLE001 - the main pipeline also goes on without vocals
+            logger.warning(f"Window {index}: vocal analysis failed: {e}")
+        reading = xai.predict_window(features, vocals, window_sec=end - start)
+        if reading is None:
+            return {**base, "state": "failed"}
+        return {
+            **base,
+            "state": "ok",
+            "probability": round(reading.probability, 4),
+            "isAi": reading.is_ai,
+            "margin": reading.margin,
+            "hasVocals": bool(vocals and vocals.has_vocals),
+            "reasons": [
+                {
+                    "name": c.name,
+                    "label": c.label,
+                    "labelEn": c.label_en,
+                    "direction": c.direction,
+                    "shapValue": round(c.shap_value, 4),
+                }
+                for c in reading.reasons
+            ],
+        }
+    except Exception as e:  # noqa: BLE001 - one bad window must not sink the scan
+        logger.warning(f"Window {index} ({start:.0f}-{end:.0f}s) failed: {e}")
+        return {**base, "state": "failed"}
+
+
+def _summarise(segments: List[Dict[str, Any]]) -> Dict[str, Any]:
+    scored = [s for s in segments if s["state"] == "ok"]
+    seconds = sum(s["end"] - s["start"] for s in scored)
+    if not scored or seconds <= 0:
+        return {
+            "scoredCount": 0, "flaggedCount": 0, "scoredSec": 0.0,
+            "aiShare": 0.0, "meanProbability": 0.0, "maxProbability": 0.0, "peakIndex": None,
+        }
+    flagged = [s for s in scored if s["isAi"]]
+    peak = max(scored, key=lambda s: s["probability"])
+    return {
+        "scoredCount": len(scored),
+        "flaggedCount": len(flagged),
+        "scoredSec": round(seconds, 1),
+        "aiShare": round(sum(s["end"] - s["start"] for s in flagged) / seconds, 3),
+        "meanProbability": round(sum(s["probability"] * (s["end"] - s["start"]) for s in scored) / seconds, 4),
+        "maxProbability": peak["probability"],
+        "peakIndex": peak["index"],
+    }
+
+
+def scan_track(
+    y: np.ndarray,
+    sr: int,
+    xai: XAIInferenceService,
+    *,
+    total_sec: float,
+    on_progress: Optional[Callable[[int, int], None]] = None,
+    cancelled: Optional[Callable[[], bool]] = None,
+    budget_sec: float = SCAN_BUDGET_SEC,
+) -> Dict[str, Any]:
+    """Read every window of `y` and return the timeline payload (blocking)."""
+    duration = y.size / sr
+    windows = plan_windows(duration)
+    started = time.monotonic()
+    lock = threading.Lock()
+    finished = 0
+
+    def work(item: Tuple[int, Tuple[float, float]]) -> Dict[str, Any]:
+        nonlocal finished
+        index, (start, end) = item
+        if (cancelled is not None and cancelled()) or time.monotonic() - started > budget_sec:
+            out: Dict[str, Any] = {"index": index, "start": round(start, 2), "end": round(end, 2), "state": "skipped"}
+        else:
+            out = _read_window(y, sr, index, start, end, xai)
+        with lock:
+            finished += 1
+            if on_progress is not None:
+                on_progress(finished, len(windows))
+        return out
+
+    with ThreadPoolExecutor(max_workers=min(SCAN_WORKERS, max(1, len(windows)))) as pool:
+        segments = list(pool.map(work, enumerate(windows)))
+
+    logger.info(
+        f"Timeline: {len(windows)} windows over {duration:.0f}s in {time.monotonic() - started:.1f}s, "
+        f"{sum(1 for s in segments if s['state'] == 'ok')} scored"
+    )
+    return {
+        "durationSec": round(duration, 2),
+        "totalSec": round(max(total_sec, duration), 2),
+        "truncated": total_sec - duration > 1.0,
+        "threshold": round(float(xai.threshold), 6),
+        "peaks": waveform_peaks(y),
+        "segments": segments,
+        "summary": _summarise(segments),
+    }

tests/test_timeline.py

@@ -0,0 +1,208 @@
+"""Whole-track scan: window plan, waveform peaks and the scan loop.
+
+The classifier is a stub here; only the audio maths and the bookkeeping run.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+from typing import List, Optional
+
+import numpy as np
+import pytest
+
+from app.services import timeline
+from app.services.feature_extractor import AudioFeatures
+from app.services.inference_xai import FeatureContribution, WindowReading, XAIInferenceService
+
+SR = 22050
+
+
+class StubXai:
+    """Scores a window by how loud it is, so each test can steer the result."""
+
+    threshold = 0.43
+
+    def __init__(self, probabilities: Optional[List[float]] = None) -> None:
+        self._queue = list(probabilities or [])
+
+    def predict_window(self, features, vocals, *, window_sec: float, top: int = 3) -> WindowReading:
+        p = self._queue.pop(0) if self._queue else 0.8
+        reason = FeatureContribution(
+            name="spectral_flatness_mean", label="Spektral düzlük", label_en="Spectral flatness",
+            category="spectral", value=0.1, z_score=-1.2, shap_value=0.4 if p >= self.threshold else -0.4,
+            direction="towards_ai" if p >= self.threshold else "towards_human", description="",
+        )
+        return WindowReading(probability=p, is_ai=p >= self.threshold, margin=0.5, reasons=[reason])
+
+
+@pytest.fixture
+def fast_features(monkeypatch):
+    """Skip the librosa work: the scan loop is what these tests cover."""
+    monkeypatch.setattr(timeline, "extract_features_array", lambda seg, sr: object())
+    monkeypatch.setattr(timeline, "analyze_vocals_array", lambda seg, sr: None)
+
+
+def tone(seconds: float, level: float = 0.5) -> np.ndarray:
+    t = np.arange(int(seconds * SR)) / SR
+    return (level * np.sin(2 * np.pi * 220 * t)).astype(np.float32)
+
+
+# ── plan ─────────────────────────────────────────────────────────────
+
+@pytest.mark.parametrize("duration", [3.0, 29.0, 44.9, 45.0, 75.0, 214.2, 359.9])
+def test_windows_tile_the_whole_track(duration: float) -> None:
+    windows = timeline.plan_windows(duration)
+    assert windows[0][0] == 0
+    assert windows[-1][1] == pytest.approx(duration, abs=0.01)
+    for (_, end), (start, _) in zip(windows, windows[1:]):
+        assert end == pytest.approx(start, abs=0.01)
+
+
+@pytest.mark.parametrize("duration", [45.0, 75.0, 214.2, 359.9])
+def test_window_length_stays_near_thirty_seconds(duration: float) -> None:
+    for start, end in timeline.plan_windows(duration):
+        assert 20.0 <= end - start <= 45.0
+
+
+def test_nothing_to_plan_for_empty_audio() -> None:
+    assert timeline.plan_windows(0) == []
+
+
+# ── peaks ────────────────────────────────────────────────────────────
+
+def test_peaks_are_normalised_and_bounded() -> None:
+    peaks = timeline.waveform_peaks(tone(10.0, 0.2), points=100)
+    assert len(peaks) == 100
+    assert max(peaks) == 1.0
+    assert min(peaks) >= 0.0
+
+
+def test_peaks_of_silence_are_zero() -> None:
+    assert set(timeline.waveform_peaks(np.zeros(SR, dtype=np.float32), points=10)) == {0.0}
+
+
+def test_peaks_of_a_very_short_clip_do_not_overrun() -> None:
+    assert len(timeline.waveform_peaks(np.ones(5, dtype=np.float32), points=600)) == 5
+
+
+# ── scan ─────────────────────────────────────────────────────────────
+
+def test_scan_scores_every_window_and_summarises(fast_features) -> None:
+    y = tone(90.0)  # three windows of 30 s
+    out = timeline.scan_track(y, SR, StubXai([0.9, 0.1, 0.7]), total_sec=90.0)
+
+    assert [s["state"] for s in out["segments"]] == ["ok", "ok", "ok"]
+    assert [s["isAi"] for s in out["segments"]] == [True, False, True]
+    summary = out["summary"]
+    assert summary["scoredCount"] == 3
+    assert summary["flaggedCount"] == 2
+    assert summary["aiShare"] == pytest.approx(2 / 3, abs=0.01)
+    assert summary["peakIndex"] == 0
+    assert summary["maxProbability"] == 0.9
+    assert out["truncated"] is False
+    assert out["threshold"] == pytest.approx(0.43)
+    assert len(out["peaks"]) > 0
+    assert out["segments"][0]["reasons"][0]["direction"] == "towards_ai"
+
+
+def test_scan_keeps_window_order_with_parallel_workers(fast_features, monkeypatch) -> None:
+    monkeypatch.setattr(timeline, "SCAN_WORKERS", 4)
+    out = timeline.scan_track(tone(300.0), SR, StubXai(), total_sec=300.0)
+    assert [s["index"] for s in out["segments"]] == list(range(10))
+    starts = [s["start"] for s in out["segments"]]
+    assert starts == sorted(starts)
+
+
+def test_a_silent_window_is_not_scored(fast_features) -> None:
+    y = np.concatenate([tone(30.0), np.zeros(30 * SR, dtype=np.float32)])
+    out = timeline.scan_track(y, SR, StubXai([0.8]), total_sec=60.0)
+    assert [s["state"] for s in out["segments"]] == ["ok", "silent"]
+    assert out["summary"]["scoredCount"] == 1
+    assert out["summary"]["scoredSec"] == pytest.approx(30.0, abs=0.1)
+
+
+def test_windows_past_the_budget_are_skipped(fast_features) -> None:
+    out = timeline.scan_track(tone(90.0), SR, StubXai(), total_sec=90.0, budget_sec=-1.0)
+    assert {s["state"] for s in out["segments"]} == {"skipped"}
+    assert out["summary"]["scoredCount"] == 0
+    assert out["summary"]["peakIndex"] is None
+
+
+def test_cancelling_skips_the_remaining_windows(fast_features) -> None:
+    out = timeline.scan_track(tone(90.0), SR, StubXai(), total_sec=90.0, cancelled=lambda: True)
+    assert {s["state"] for s in out["segments"]} == {"skipped"}
+
+
+def test_a_long_track_is_flagged_as_truncated(fast_features) -> None:
+    out = timeline.scan_track(tone(30.0), SR, StubXai(), total_sec=500.0)
+    assert out["truncated"] is True
+    assert out["totalSec"] == 500.0
+    assert out["durationSec"] == pytest.approx(30.0, abs=0.05)
+
+
+def test_a_failing_window_does_not_sink_the_scan(fast_features, monkeypatch) -> None:
+    calls = {"n": 0}
+
+    def flaky(seg, sr):
+        calls["n"] += 1
+        if calls["n"] == 2:
+            raise RuntimeError("boom")
+        return object()
+
+    monkeypatch.setattr(timeline, "extract_features_array", flaky)
+    monkeypatch.setattr(timeline, "SCAN_WORKERS", 1)
+    out = timeline.scan_track(tone(90.0), SR, StubXai(), total_sec=90.0)
+    assert [s["state"] for s in out["segments"]] == ["ok", "failed", "ok"]
+    assert out["summary"]["scoredCount"] == 2
+
+
+def test_progress_reports_every_window(fast_features) -> None:
+    seen: List[tuple] = []
+    timeline.scan_track(tone(90.0), SR, StubXai(), total_sec=90.0, on_progress=lambda d, n: seen.append((d, n)))
+    assert sorted(seen) == [(1, 3), (2, 3), (3, 3)]
+
+
+def test_real_audio_features_run_on_a_short_window() -> None:
+    """The librosa path, with only the classifier stubbed."""
+    rng = np.random.default_rng(3)
+    y = (tone(4.0, 0.3) + 0.02 * rng.standard_normal(4 * SR)).astype(np.float32)
+    out = timeline.scan_track(y, SR, StubXai([0.2]), total_sec=4.0)
+    assert out["segments"][0]["state"] == "ok"
+    assert out["segments"][0]["probability"] == 0.2
+
+
+def test_window_beat_count_is_scaled_to_the_training_clip() -> None:
+    """A 30 s window has about half the beats of a 60 s clip; the model expects 60 s."""
+    seen = {}
+
+    class Scaler:
+        def transform(self, x):
+            seen["x"] = x.copy()
+            return x
+
+    class Model:
+        def predict_proba(self, x):
+            return np.array([[0.4, 0.6]])
+
+    svc = XAIInferenceService.__new__(XAIInferenceService)
+    svc.available = True
+    svc.threshold = 0.43
+    svc.feature_cols = ["beat_count", "tempo_bpm"]
+    svc.feature_stats = {}
+    svc.shap_explainer = None
+    svc.scaler = Scaler()
+    svc.model = Model()
+
+    features = AudioFeatures(**{f.name: 0 for f in dataclasses.fields(AudioFeatures)})
+    features.beat_count = 25
+    features.tempo_bpm = 120.0
+
+    reading = svc.predict_window(features, None, window_sec=30.0)
+
+    assert seen["x"][0, 0] == 50.0
+    assert seen["x"][0, 1] == 120.0
+    assert reading is not None
+    assert reading.probability == pytest.approx(0.6)
+    assert reading.is_ai is True
+    assert reading.reasons == []