feat(douanier): modular ears building blocks — /features + /analyze/samples (#38)

Two new CPU-native, torch-free, cached endpoints on the audio sub-API — the
"interesting value points" for hexa beyond emotion, each the same shape/pattern
as /analyze/emotion (upload → content-address → cache → compute).

- POST /features  → the ~35-dim audio feature stack (spectral moments, MFCCs,
  chroma/key, envelope/attack-decay, + rhythm/tempo). scope `features`.
- POST /analyze/samples → per-sample EDA + role (percs|bass|melodic|tops|atmos)
  decided by the MEASURED spectrum (centroid + band energy), never the name; an
  optional ?name= only disambiguates breaks/drums (feedback_mastering_eda). scope `samples`.

Engine: engines/feats.py is SELF-CONTAINED (vendored DSP), like ears_light —
the deployed container holds only armada/api/, not armada/tide-table/, so it can
NOT import sample_features/audio_lens at runtime. It mirrors their algorithms and
fixes the librosa-0.11 tempo bug (feature.tempo, not the removed
feature.rhythm.tempo that silently dropped tempo in the tide-table original).

Wiring: a shared _cached() helper now backs all three analyze routes (DRY);
healthz advertises the engines map; both routes are scope-gated via the gateway
X-Scopes. Validated in the torch-free container: /features 200 (41 features,
tempo 107.7, key=Amaj), /analyze/samples role-by-measurement, 403 scope gate,
X-Douanier-Cache hit on repeat. 40/40 tests (added test_feats.py + endpoint cases).

armada/api/README.md

@@ -53,7 +53,12 @@ Endpoints (internal root paths shown; public = `/audio/v1` + path):
 - `GET  /metrics` — Prometheus text for the erable scraper (unauthenticated)
 - `GET  /me` — echo the caller's gateway identity (any authenticated tenant)
 - `POST /analyze/emotion` — `multipart` audio → `{valence, arousal, top…}` (scope `emotion`)
+- `POST /features` — audio → the ~35-dim feature stack (spectral/MFCC/chroma/key/envelope + rhythm); `?rhythm=` (scope `features`)
+- `POST /analyze/samples` — one-shot/loop → per-sample EDA + MEASURED role (percs|bass|melodic|tops|atmos); `?name=` only disambiguates breaks/drums (scope `samples`)
 - `POST /separate` — stem separation; **503 `compute_unavailable`** until a GPU runner (scope `separate`)
+
+All three analyze routes are **torch-free** (librosa) and **content-addressed
+cached** (#26): the same clip recomputes once, then serves `X-Douanier-Cache: hit`.
 - `GET  /docs` — Swagger UI
 
 ```bash

armada/api/app.py

@@ -18,7 +18,7 @@ from fastapi.middleware.cors import CORSMiddleware
 import auth
 import cache
 import config
-from engines import ears
+from engines import ears, feats
 
 app = FastAPI(
     title="Douanier — the `audio` sub-API of the nech.pl platform",
@@ -99,6 +99,8 @@ def healthz():
         "status": "ok",
         "engines": {
             "emotion": {"selected": config.EMOTION_ENGINE, "available": ok, "hint": hint},
+            "features": {"available": ok, "hint": "light feature stack (librosa, no torch)"},
+            "samples": {"available": ok, "hint": "measured sample EDA + role (librosa, no torch)"},
             "separate": {"available": False, "hint": "no CPU path on erable — GPU runner pending"},
         },
     }
@@ -142,37 +144,71 @@ def whoami(principal: auth.Principal = Depends(require_scope(None))):
             "expires_at": principal.expires_at}
 
 
-# ── #36 walking-skeleton endpoint: synchronous emotion read ─────────────────────
-@v1.post("/analyze/emotion")
-async def analyze_emotion(response: Response, file: UploadFile = File(...),
-                          _p: auth.Principal = Depends(require_scope("emotion"))):
-    """Upload a short audio clip → valence/arousal + top emotions (CLAP).
-
-    Content-addressed cache (#26): the same audio returns instantly on a repeat
-    (X-Douanier-Cache: hit) instead of paying the ~CLAP compute again. Still
-    synchronous on a miss; #28 layers the async job queue under this.
-    """
+# ── shared analyze flow: upload → content-address → cache → compute (#26) ───────
+async def _read_upload(file: UploadFile) -> bytes:
     data = await file.read()
     limit = config.MAX_UPLOAD_MB * 1024 * 1024
     if len(data) > limit:
         raise HTTPException(413, f"file too large (> {config.MAX_UPLOAD_MB} MB)")
-    suffix = Path(file.filename or "clip").suffix or ".wav"
+    return data
+
+
+def _cached(kind: str, params: dict, data: bytes, filename: str, compute, response: Response):
+    """Content-address the bytes, serve a cache hit instantly (X-Douanier-Cache:
+    hit), else run `compute(path)` and cache it. Returns (content_id, result).
+    The same engine call is paid ONCE per (content_id, params) — the margin lever."""
+    suffix = Path(filename or "clip").suffix or ".wav"
     with tempfile.NamedTemporaryFile(suffix=suffix, delete=True) as tmp:
         tmp.write(data)
         tmp.flush()
         cid = cache.content_id(tmp.name)
-        params = {"engine": config.EMOTION_ENGINE}
-        hit = cache.get(cid, "emotion", params)
+        hit = cache.get(cid, kind, params)
         if hit:
             response.headers["X-Douanier-Cache"] = "hit"
-            return {"filename": file.filename, "content_id": cid, "emotion": hit["result"]}
+            return cid, hit["result"]
         try:
-            read = ears.emotion_read(tmp.name)
+            result = compute(tmp.name)
         except Exception as e:
-            raise HTTPException(422, f"emotion analysis failed: {e}")
-    cache.put(cid, "emotion", result=read, params=params)
+            raise HTTPException(422, f"{kind} analysis failed: {e}")
+    cache.put(cid, kind, result=result, params=params)
     response.headers["X-Douanier-Cache"] = "miss"
-    return {"filename": file.filename, "content_id": cid, "emotion": read}
+    return cid, result
+
+
+@v1.post("/analyze/emotion")
+async def analyze_emotion(response: Response, file: UploadFile = File(...),
+                          _p: auth.Principal = Depends(require_scope("emotion"))):
+    """Upload a short audio clip → valence/arousal + top emotions. Cached (#26):
+    the same audio returns instantly on a repeat instead of recomputing."""
+    data = await _read_upload(file)
+    params = {"engine": config.EMOTION_ENGINE}
+    cid, result = _cached("emotion", params, data, file.filename, ears.emotion_read, response)
+    return {"filename": file.filename, "content_id": cid, "emotion": result}
+
+
+@v1.post("/features")
+async def features(response: Response, file: UploadFile = File(...), rhythm: bool = True,
+                   _p: auth.Principal = Depends(require_scope("features"))):
+    """Upload a clip → the ~35-dim audio feature stack (spectral moments, MFCCs,
+    chroma/key, envelope, + rhythm when `rhythm=true`). Torch-free, cached."""
+    data = await _read_upload(file)
+    params = {"engine": "light", "rhythm": rhythm}
+    cid, result = _cached("features", params, data, file.filename,
+                          lambda p: feats.features(p, rhythm=rhythm), response)
+    return {"filename": file.filename, "content_id": cid, **result}
+
+
+@v1.post("/analyze/samples")
+async def analyze_samples(response: Response, file: UploadFile = File(...), name: str = "",
+                          _p: auth.Principal = Depends(require_scope("samples"))):
+    """Upload a one-shot/loop → per-sample EDA + MEASURED role (percs|bass|melodic|
+    tops|atmos) from the spectrum, never the name. Optional `name` only
+    disambiguates breaks/drums. Torch-free, cached."""
+    data = await _read_upload(file)
+    params = {"engine": "light", "name": name}
+    cid, result = _cached("samples", params, data, file.filename,
+                          lambda p: feats.sample_profile(p, name=name), response)
+    return {"filename": file.filename, "content_id": cid, **result}
 
 
 # ── #37 separation seam — no CPU path on erable; env-selected GPU backend later ──

armada/api/engines/feats.py

+"""Light feature + sample-analysis engines — CPU, NO torch.
+
+Two modular building blocks for hexa, the same shape/pattern as ears_light:
+  • features(path)        → the ~35-dim audio feature stack (/features)
+  • sample_profile(path)  → per-sample EDA + measured role (/analyze/samples)
+
+SELF-CONTAINED on purpose: the deployed container holds only armada/api/ (not
+armada/tide-table/), so these VENDOR the DSP rather than importing it. They
+mirror the canonical tide-table implementations — sample_features.features() and
+audio_lens.profile()/classify_family() — and test_feats cross-checks the feature
+KEYS against tide-table when it's importable, so drift is caught.
+
+Mastering discipline (feedback_mastering_eda): role/register is decided by the
+MEASURED spectrum (centroid + band energy), never inferred from a name. An
+optional `name` only disambiguates what spectrum can't (a break is broadband; a
+kick and a sub are both sub-band).
+"""
+import numpy as np
+
+SR = 22050                                   # MIR-standard; plenty for timbre/role
+MAX_S = 30.0                                 # cap clip length (loops/long tails)
+
+# Krumhansl-Schmuckler major/minor key profiles (chroma → key estimate)
+_KS_MAJ = np.array([6.35, 2.23, 3.48, 2.33, 4.38, 4.09, 2.52, 5.19, 2.39, 3.66, 2.29, 2.88])
+_KS_MIN = np.array([6.33, 2.68, 3.52, 5.38, 2.60, 3.53, 2.54, 4.75, 3.98, 2.69, 3.34, 3.17])
+_NOTES = ["C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "A#", "B"]
+
+# feature hierarchy (mirrors sample_features.FEATURE_TIERS — robust L0 → derived L1)
+FEATURE_TIERS = {
+    "L0": ["rms_db", "peak_db", "crest_db", "zcr", "spectral_centroid", "duration_s"],
+    "L1": ["log_attack_time", "temporal_centroid", "decay_slope_db_s", "sustain_ratio",
+           "spectral_spread", "spectral_skew", "spectral_kurtosis", "spectral_rolloff",
+           "spectral_flatness", "spectral_bandwidth", "spectral_flux", "spectral_contrast",
+           "chroma_entropy", "key_strength", "f0_median", "hnr", "pct_percussive",
+           "onset_rate", "tempo_bpm", "pulse_clarity"] + [f"mfcc_{i}" for i in range(13)],
+}
+
+# band edges (Hz) for sample band-energy %, mirrors audio_lens.BANDS
+BANDS = [("<80", 0, 80), ("80-150", 80, 150), ("150-500", 150, 500),
+         ("500-2k", 500, 2000), ("2k+", 2000, SR / 2)]
+
+# role hints (mirror audio_lens) — identity only disambiguates what spectrum can't
+_BREAK_HINTS = ("break", "jungle", "amen", "org_jungle")
+_PERC_EXACT = {"kick", "bd", "snare", "sn", "sd", "dr", "drum", "clap", "cp",
+               "rim", "rs", "hh", "hat", "hats", "cym", "cy", "tom", "perc",
+               "crash", "ride", "shaker", "conga", "bongo"}
+_PERC_PREFIX = ("kick", "snare", "clap", "hat", "perc", "rample", "h2ogm",
+                "reverbkick", "808kick", "909")
+_BASS_HINTS = ("bass", "sub", "808", "reese", "moog", "fbass", "acid", "wobble")
+
+
+def _load(audio_path, max_seconds, sr=SR):
+    import librosa
+    y, _ = librosa.load(str(audio_path), sr=sr, mono=True, duration=max_seconds)
+    if y.size < sr // 20:                    # need ≥ 50 ms
+        raise ValueError("clip too short / undecodable (need ≥ 50 ms of audio)")
+    return y
+
+
+def _env_features(y, sr):
+    """Temporal envelope descriptors — the kick↔bass discriminators (Peeters)."""
+    import librosa
+    rms = librosa.feature.rms(y=y, frame_length=1024, hop_length=256)[0]
+    t = librosa.times_like(rms, sr=sr, hop_length=256)
+    if rms.sum() < 1e-9 or len(rms) < 3:
+        return {}
+    pk = int(np.argmax(rms))
+    peak = rms[pk] + 1e-9
+    attack_t = max(t[pk], 1e-3)
+    tempo_cen = float((t * rms).sum() / (rms.sum() + 1e-9))
+    tail = rms[pk:]
+    if len(tail) > 2:
+        tt = t[pk:] - t[pk]
+        decay = float(np.polyfit(tt, 20 * np.log10(tail + 1e-9), 1)[0])
+    else:
+        decay = 0.0
+    return {"log_attack_time": float(np.log10(attack_t)),
+            "temporal_centroid": tempo_cen,
+            "decay_slope_db_s": decay,
+            "sustain_ratio": float(np.mean(rms[pk:]) / peak),
+            "duration_s": float(len(y) / sr)}
+
+
+def features(audio_path, max_seconds=MAX_S, rhythm=True) -> dict:
+    """Flat dict of ~35 features (+ tiers + engine tag). Robust to short/silent
+    clips (a missing feature is omitted, not faked)."""
+    import librosa
+    y = _load(audio_path, max_seconds)
+    f = {}
+    # ── time domain ──
+    rms = np.sqrt(np.mean(y ** 2)) + 1e-9
+    peak = np.abs(y).max() + 1e-9
+    f["rms_db"] = float(20 * np.log10(rms))
+    f["peak_db"] = float(20 * np.log10(peak))
+    f["crest_db"] = float(20 * np.log10(peak / rms))
+    f["zcr"] = float(librosa.feature.zero_crossing_rate(y)[0].mean())
+    f.update(_env_features(y, SR))
+    # ── spectral (moments + shape) ──
+    S = np.abs(librosa.stft(y, n_fft=2048, hop_length=512)) ** 2
+    freqs = librosa.fft_frequencies(sr=SR, n_fft=2048)
+    Sm = S.mean(axis=1)
+    tot = Sm.sum() + 1e-12
+    cen = float((freqs * Sm).sum() / tot)
+    spread = float(np.sqrt(((freqs - cen) ** 2 * Sm).sum() / tot))
+    f["spectral_centroid"] = cen
+    f["spectral_spread"] = spread
+    f["spectral_skew"] = float(((freqs - cen) ** 3 * Sm).sum() / (tot * spread ** 3 + 1e-9))
+    f["spectral_kurtosis"] = float(((freqs - cen) ** 4 * Sm).sum() / (tot * spread ** 4 + 1e-9))
+    f["spectral_rolloff"] = float(librosa.feature.spectral_rolloff(y=y, sr=SR).mean())
+    f["spectral_flatness"] = float(librosa.feature.spectral_flatness(y=y).mean())
+    f["spectral_bandwidth"] = float(librosa.feature.spectral_bandwidth(y=y, sr=SR).mean())
+    f["spectral_flux"] = float(np.mean(np.diff(S, axis=1).clip(min=0).sum(axis=0)))
+    f["spectral_contrast"] = float(librosa.feature.spectral_contrast(y=y, sr=SR).mean())
+    # ── cepstral timbre ──
+    mfcc = librosa.feature.mfcc(y=y, sr=SR, n_mfcc=13).mean(axis=1)
+    for i, v in enumerate(mfcc):
+        f[f"mfcc_{i}"] = float(v)
+    # ── harmonic / percussive ──
+    try:
+        yh, yp = librosa.effects.hpss(y)
+        eh, ep = float(np.sum(yh ** 2)), float(np.sum(yp ** 2))
+        f["pct_percussive"] = float(ep / (eh + ep + 1e-9))
+        f["hnr"] = float(eh / (ep + 1e-9))
+    except Exception:
+        pass
+    # ── pitch / key ──
+    try:
+        chroma = librosa.feature.chroma_cqt(y=y, sr=SR).mean(axis=1)
+        cs = chroma / (chroma.sum() + 1e-9)
+        f["chroma_entropy"] = float(-(cs * np.log2(cs + 1e-12)).sum())
+        cor = [(np.corrcoef(np.roll(_KS_MAJ, k), chroma)[0, 1], _NOTES[k], "maj") for k in range(12)]
+        cor += [(np.corrcoef(np.roll(_KS_MIN, k), chroma)[0, 1], _NOTES[k], "min") for k in range(12)]
+        best = max(cor, key=lambda x: (x[0] if np.isfinite(x[0]) else -9))
+        f["key"], f["key_mode"], f["key_strength"] = best[1], best[2], float(best[0])
+    except Exception:
+        pass
+    try:
+        f0 = librosa.yin(y, fmin=40, fmax=2000, sr=SR)
+        f0 = f0[np.isfinite(f0)]
+        if f0.size:
+            f["f0_median"] = float(np.median(f0))
+    except Exception:
+        pass
+    # ── rhythm (loops/takes) — librosa 0.11: feature.tempo, NOT feature.rhythm.tempo ──
+    if rhythm:
+        try:
+            onset_env = librosa.onset.onset_strength(y=y, sr=SR)
+            onsets = librosa.onset.onset_detect(onset_envelope=onset_env, sr=SR)
+            dur = len(y) / SR
+            f["onset_rate"] = float(len(onsets) / dur) if dur else 0.0
+            tempo = np.atleast_1d(librosa.feature.tempo(onset_envelope=onset_env, sr=SR))
+            f["tempo_bpm"] = float(tempo[0]) if tempo.size else 0.0
+            f["pulse_clarity"] = float(np.max(librosa.autocorrelate(onset_env)) /
+                                       (np.sum(onset_env ** 2) + 1e-9))
+        except Exception:
+            pass
+    return {"features": f, "tiers": FEATURE_TIERS, "sr": SR, "engine": "light"}
+
+
+# ── sample analysis (band energy + measured role) ──────────────────────────────
+def _profile(y):
+    """Band-energy %, spectral centroid, rms/peak — the core 'what is this sound'."""
+    w = np.hanning(len(y))
+    S = np.abs(np.fft.rfft(y * w)) ** 2
+    fr = np.fft.rfftfreq(len(y), 1 / SR)
+    tot = S.sum() + 1e-20
+    bands = {name: float(100 * S[(fr >= lo) & (fr < hi)].sum() / tot) for name, lo, hi in BANDS}
+    return {"rms_db": float(20 * np.log10(np.sqrt(np.mean(y ** 2)) + 1e-9)),
+            "peak_db": float(20 * np.log10(np.abs(y).max() + 1e-9)),
+            "centroid": float((fr * S).sum() / tot), "bands": bands}
+
+
+def classify_family(name, prof):
+    """(family_key, centroid|None), measurement-aware. Families: percs|bass|melodic|tops|atmos."""
+    s = (name or "").lower()
+    cen = prof["centroid"] if prof else None
+    if any(h in s for h in _BREAK_HINTS):
+        return "tops", cen
+    if s in _PERC_EXACT or any(s.startswith(h) for h in _PERC_PREFIX):
+        return "percs", cen
+    bass_name = any(h in s for h in _BASS_HINTS)
+    if cen is None:
+        return ("bass" if bass_name else "melodic"), None
+    if cen < 180:
+        return "bass", cen
+    if cen < 1400:
+        return ("bass" if bass_name else "melodic"), cen
+    if cen < 3500:
+        return "melodic", cen
+    return "atmos", cen
+
+
+def _time_activity(y, sr):
+    """Fraction of the clip above -40 dB of its peak — sustained vs one-shot."""
+    import librosa
+    rms = librosa.feature.rms(y=y, frame_length=1024, hop_length=256)[0]
+    if rms.size == 0:
+        return 0.0
+    thresh = rms.max() * (10 ** (-40 / 20))
+    return float(np.mean(rms > thresh))
+
+
+def sample_profile(audio_path, name="", max_seconds=MAX_S) -> dict:
+    """Per-sample EDA the way mastering needs it: measured spectrum → role, never
+    inferred from the name. `name` (optional) only disambiguates breaks/drums."""
+    y = _load(audio_path, max_seconds)
+    prof = _profile(y)
+    family, cen = classify_family(name, prof)
+    return {
+        "profile": prof,
+        "role": family,
+        "centroid": cen,
+        "broadband": bool(prof["bands"]["150-500"] + prof["bands"]["500-2k"]
+                          + prof["bands"]["2k+"] > 18 and prof["centroid"] > 140),
+        "time_activity": _time_activity(y, SR),
+        "name_hint": name or None,
+        "sr": SR,
+        "engine": "light",
+    }

armada/api/tests/test_feats.py

+"""Light feature + sample-analysis engine tests (#38) — torch-free, real DSP.
+
+Synth clips with known properties so the asserts are about MEASUREMENT, not a
+fixture: a low sine reads bass register, a bright noise reads high register, the
+feature stack carries the L0 floor + a correct tempo. conftest sets up paths.
+
+    cd armada/api && ~/.virtualenvs/douanier/bin/python -m pytest tests/test_feats.py -q
+"""
+import sys
+import tempfile
+from pathlib import Path
+
+import numpy as np
+import soundfile as sf
+
+from engines import feats
+
+SR = 22050
+
+
+def _write(y):
+    p = Path(tempfile.mktemp(suffix=".wav"))
+    sf.write(str(p), y.astype("float32"), SR)
+    return str(p)
+
+
+def _sine(freqs, secs=2.0):
+    t = np.linspace(0, secs, int(SR * secs), endpoint=False)
+    y = sum(np.sin(2 * np.pi * f * t) for f in freqs) / len(freqs)
+    return _write(0.3 * y)
+
+
+def _noise(secs=2.0, highpass=False):
+    rng = np.random.default_rng(0)
+    y = rng.standard_normal(int(SR * secs))
+    if highpass:                      # crude high-emphasis → bright, high centroid
+        y = np.diff(y, prepend=0.0)
+    return _write(0.2 * y / (np.abs(y).max() + 1e-9))
+
+
+def test_features_has_L0_floor_and_tempo():
+    r = feats.features(_sine([220, 277, 330]), rhythm=True)
+    f = r["features"]
+    # the robust L0 tier must always be present
+    for k in feats.FEATURE_TIERS["L0"]:
+        assert k in f, f"missing L0 feature {k}"
+    assert r["engine"] == "light" and r["sr"] == SR
+    # tempo came through the librosa 0.11 path (feature.tempo, not rhythm.tempo)
+    assert "tempo_bpm" in f and f["tempo_bpm"] > 0
+
+
+def test_features_detects_major_key():
+    f = feats.features(_sine([220, 277.18, 329.63]), rhythm=False)["features"]
+    assert f.get("key") == "A" and f.get("key_mode") == "maj"
+
+
+def test_sample_role_is_measured_not_named():
+    # a 60 Hz sine is bass by SPECTRUM even if we (mis)name it "lead"
+    low = feats.sample_profile(_sine([60]), name="lead_synth")
+    assert low["role"] == "bass" and low["centroid"] < 180
+    # a bright noise sits in the high register
+    hi = feats.sample_profile(_noise(highpass=True), name="")
+    assert hi["role"] in {"melodic", "atmos"} and hi["centroid"] > low["centroid"]
+
+
+def test_sample_bands_sum_to_100():
+    sp = feats.sample_profile(_sine([220, 440]), name="")
+    total = sum(sp["profile"]["bands"].values())
+    assert abs(total - 100.0) < 0.5
+    assert "time_activity" in sp and 0.0 <= sp["time_activity"] <= 1.0
+
+
+def test_engine_pulls_no_torch():
+    feats.features(_sine([330]), rhythm=False)
+    feats.sample_profile(_sine([330]), name="")
+    assert "torch" not in sys.modules     # the light path must never import torch

armada/api/tests/test_smoke.py

@@ -98,6 +98,46 @@ def test_emotion_too_large(monkeypatch):
     assert r.status_code == 413
 
 
+def _wav_bytes(freqs=(220, 277, 330), secs=1.5):
+    import io
+    import numpy as np
+    import soundfile as sf
+    t = np.linspace(0, secs, int(22050 * secs), endpoint=False)
+    y = (0.3 * sum(np.sin(2 * np.pi * f * t) for f in freqs) / len(freqs)).astype("float32")
+    buf = io.BytesIO()
+    sf.write(buf, y, 22050, format="WAV")
+    return buf.getvalue()
+
+
+def test_features_endpoint_and_cache():
+    files = {"file": ("c.wav", _wav_bytes(), "audio/wav")}
+    r1 = client.post("/features", files=files, headers={"X-Tenant": "hexa", "X-Scopes": "features"})
+    assert r1.status_code == 200
+    assert r1.headers["X-Douanier-Cache"] == "miss"
+    body = r1.json()
+    assert "features" in body and body["engine"] == "light"
+    assert "spectral_centroid" in body["features"]
+    r2 = client.post("/features", files={"file": ("c.wav", _wav_bytes(), "audio/wav")},
+                     headers={"X-Tenant": "hexa", "X-Scopes": "features"})
+    assert r2.headers["X-Douanier-Cache"] == "hit"
+
+
+def test_features_scope_enforced():
+    r = client.post("/features", files={"file": ("c.wav", b"RIFF....", "audio/wav")},
+                    headers={"X-Tenant": "hexa", "X-Scopes": "emotion"})
+    assert r.status_code == 403
+
+
+def test_samples_endpoint_measures_role():
+    files = {"file": ("c.wav", _wav_bytes(freqs=(60,)), "audio/wav")}
+    r = client.post("/analyze/samples?name=lead", files=files,
+                    headers={"X-Tenant": "hexa", "X-Scopes": "samples"})
+    assert r.status_code == 200
+    body = r.json()
+    # 60 Hz sine → bass by measurement, despite the "lead" name hint
+    assert body["role"] == "bass" and body["engine"] == "light"
+
+
 def test_separate_is_503_until_gpu():
     r = client.post("/separate", headers={"X-Tenant": "hexa", "X-Scopes": "separate"})
     assert r.status_code == 503