feat(audio): full-track streaming analysis — accept big uploads instead of 413

Producers upload whole tracks, not 60s clips. Before, a large file was either
413-rejected at the edge or buffered wholesale in RAM (upload read + content
hash + engine decode each copied the whole file). Now ingest, hashing and the
signal/loudness engines all stream in bounded windows, so peak memory is
O(window) not O(file) and a big upload can't OOM the shared box.

- engines/streaming.py: probe() + windows() — bounded window iterator over the
  whole file at native SR (soundfile streams WAV/FLAC/OGG/MP3, exact frames).
- onsets/waveform/spectrum: full-track. waveform/spectrum aggregate EXACTLY
  (fixed output grid filled by a streaming reduction); onsets concatenates
  per-window onset envelopes (tiny) for near-exact global tempo/hits.
- loudness: full-track; peak/true-peak/RMS/crest exact, integrated LUFS
  energy-weighted across windows (pyloudnorm gating can't merge chunks exactly)
  and flagged approximate:true / lufs_method:chunked.
- clip-tier (emotion/features/samples/analyze/grade/naming) stay head-analysis
  (a 5-min track's 'sample role' is meaningless) but every response now carries
  an honest source{duration_s,sr,channels,size_bytes} block.
- app ingest streams the upload to disk (never one big read); size ceiling read
  per-request. content_id streams the decode-to-hash, byte-identical to the old
  whole-file hash so the deployed cache stays warm. Upload cap 40->100MB to
  match the gateway.
- tests: test_fulltrack.py proves >60s coverage, content_id stability, exact
  streaming reductions, contiguous windowing. Full suite green (parity w/ base).

armada/api/app.py

@@ -23,7 +23,7 @@ import cache
 import config
 import jobs as jobslib
 import scopes as scopelib
-from engines import ears, feats, signal
+from engines import ears, feats, signal, streaming
 from engines import loudness as loudness_eng
 from engines import grade as grade_eng    # VENDORED copy of the Foundry grader (see engines/grade.py)
 from engines import naming as naming_eng  # VENDORED copy of the Foundry namer (see engines/naming.py)
@@ -278,29 +278,61 @@ def get_artifact(content_id: str, name: str,
 
 
 # ── shared analyze flow: upload → content-address → cache → compute (#26) ───────
-async def _read_upload(file: UploadFile) -> bytes:
-    data = await file.read()
+# SAFETY (#7): the upload is STREAMED to a temp file in bounded chunks — never
+# `await file.read()` into one big `bytes` (a 100 MB body would spike ~100 MB
+# resident, and the content-hash + engines add their own copies). Peak memory
+# stays O(chunk), so a large upload can't OOM the shared erable box. The size
+# ceiling is enforced mid-stream (read at request time, so tests/ops can retune
+# config.MAX_UPLOAD_MB live).
+async def _spool(file: UploadFile) -> Path:
     limit = config.MAX_UPLOAD_MB * 1024 * 1024
-    if len(data) > limit:
-        raise HTTPException(413, f"file too large (> {config.MAX_UPLOAD_MB} MB)")
-    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-Nech-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)
+    suffix = Path(file.filename or "clip").suffix or ".wav"
+    tmp = tempfile.NamedTemporaryFile(suffix=suffix, delete=False)
+    path = Path(tmp.name)
+    size = 0
+    try:
+        while chunk := await file.read(1 << 20):     # 1 MB at a time
+            size += len(chunk)
+            if size > limit:
+                raise HTTPException(413, {
+                    "error": "payload_too_large",
+                    "detail": f"upload exceeds the {config.MAX_UPLOAD_MB} MB limit for this API",
+                    "docs": "https://api.nech.pl/docs",
+                })
+            tmp.write(chunk)
         tmp.flush()
-        cid = cache.content_id(tmp.name)
+    except BaseException:
+        tmp.close(); path.unlink(missing_ok=True); raise
+    else:
+        tmp.close()
+    return path
+
+
+def _source_meta(path: Path) -> dict:
+    """Honest provenance for the response: the file's TRUE duration + size, so a
+    caller always knows the full extent of what they sent — and, paired with an
+    engine's own `duration_s` / `full_track`, whether it was analyzed whole or
+    (for the clip-tier engines) only its head."""
+    meta = {"size_bytes": path.stat().st_size}
+    try:
+        p = streaming.probe(str(path))
+        meta.update(duration_s=round(p["duration_s"], 3), sr=p["sr"], channels=p["channels"])
+    except Exception:
+        pass
+    return meta
+
+
+def _cached(kind: str, params: dict, path: str, compute, response: Response):
+    """Content-address the file, serve a cache hit instantly (X-Nech-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."""
+    cid = cache.content_id(path)
     hit = cache.get(cid, kind, params)
     if hit:
         response.headers["X-Nech-Cache"] = "hit"
         return cid, hit["result"]
     try:
-            result = compute(tmp.name)
+        result = compute(path)
     except Exception as e:
         raise HTTPException(422, f"{kind} analysis failed: {e}")
     cache.put(cid, kind, result=result, params=params)
@@ -308,27 +340,39 @@ def _cached(kind: str, params: dict, data: bytes, filename: str, compute, respon
     return cid, result
 
 
+async def _run(file: UploadFile, kind: str, params: dict, compute, response: Response):
+    """Spool the upload (bounded), content-address + cache + compute, always
+    cleaning up the temp file. Returns (content_id, result, source_meta)."""
+    path = await _spool(file)
+    try:
+        source = _source_meta(path)
+        cid, result = _cached(kind, params, str(path), compute, response)
+        return cid, result, source
+    finally:
+        path.unlink(missing_ok=True)
+
+
 @v1.post("/analyze/emotion", operation_id="analyzeEmotion")
 async def analyze_emotion(response: Response, file: UploadFile = File(...),
                           _p: auth.Principal = Depends(require)):
     """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}
+    the same audio returns instantly on a repeat instead of recomputing.
+    Clip-tier: the emotion engine reads the first ~60 s (`source.duration_s`
+    reports the full length)."""
+    cid, result, source = await _run(file, "emotion", {"engine": config.EMOTION_ENGINE},
+                                     ears.emotion_read, response)
+    return {"filename": file.filename, "content_id": cid, "emotion": result, "source": source}
 
 
 @v1.post("/features", operation_id="features")
 async def features(response: Response, file: UploadFile = File(...), rhythm: bool = True,
                    _p: auth.Principal = Depends(require)):
     """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,
+    chroma/key, envelope, + rhythm when `rhythm=true`). Torch-free, cached.
+    Clip-tier: analyses the first ~30 s (`source.duration_s` is the full length)."""
+    cid, result, source = await _run(file, "features", {"engine": "light", "rhythm": rhythm},
                                      lambda p: feats.features(p, rhythm=rhythm), response)
-    return {"filename": file.filename, "content_id": cid, **result}
+    return {"filename": file.filename, "content_id": cid, **result, "source": source}
 
 
 @v1.post("/analyze/samples", operation_id="analyzeSamples")
@@ -336,49 +380,47 @@ async def analyze_samples(response: Response, file: UploadFile = File(...), name
                           _p: auth.Principal = Depends(require)):
     """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,
+    disambiguates breaks/drums. Torch-free, cached. Clip-tier: first ~30 s."""
+    cid, result, source = await _run(file, "samples", {"engine": "light", "name": name},
                                      lambda p: feats.sample_profile(p, name=name), response)
-    return {"filename": file.filename, "content_id": cid, **result}
+    return {"filename": file.filename, "content_id": cid, **result, "source": source}
 
 
 @v1.post("/onsets", operation_id="onsets")
 async def onsets(response: Response, file: UploadFile = File(...),
                  _p: auth.Principal = Depends(require)):
     """Upload audio → onset hit times (s) + tempo + onset rate. The cheapest tier
-    (no model): rhythmic hits for visual sync / slicing. Torch-free, cached."""
-    data = await _read_upload(file)
-    cid, result = _cached("onsets", {"v": 1}, data, file.filename, signal.onsets, response)
-    return {"filename": file.filename, "content_id": cid, **result}
+    (no model): rhythmic hits for visual sync / slicing. Torch-free, cached.
+    Full-track: streams the WHOLE upload (`full_track:true`), not just a head clip."""
+    cid, result, source = await _run(file, "onsets", {"v": 2}, signal.onsets, response)
+    return {"filename": file.filename, "content_id": cid, **result, "source": source}
 
 
 @v1.post("/waveform", operation_id="waveform")
 async def waveform(response: Response, file: UploadFile = File(...), bins: int = 800,
                    _p: auth.Principal = Depends(require)):
     """Upload audio → render-ready waveform: per-bin [min,max] + a 0..1 RMS energy
-    envelope (`?bins=`, ≤4000). For audio-reactive visuals. Torch-free, cached."""
-    data = await _read_upload(file)
-    cid, result = _cached("waveform", {"bins": bins, "v": 1}, data, file.filename,
+    envelope (`?bins=`, ≤4000). For audio-reactive visuals. Torch-free, cached.
+    Full-track: bins span the WHOLE track via an exact streaming reduction."""
+    cid, result, source = await _run(file, "waveform", {"bins": bins, "v": 2},
                                      lambda p: signal.waveform(p, bins=bins), response)
-    return {"filename": file.filename, "content_id": cid, **result}
+    return {"filename": file.filename, "content_id": cid, **result, "source": source}
 
 
 @v1.post("/analyze", operation_id="analyzeAll")
 async def analyze_all(response: Response, file: UploadFile = File(...),
                       _p: auth.Principal = Depends(require)):
     """One call → emotion + features + sample role for a clip (fewer round-trips
-    for hexa). Each is the same engine the dedicated routes use. Cached as a unit."""
-    data = await _read_upload(file)
+    for hexa). Each is the same engine the dedicated routes use. Cached as a unit.
+    Clip-tier composite: emotion ~60 s, features/sample ~30 s of the head."""
     params = {"engine": config.EMOTION_ENGINE, "v": 1}
 
     def _compute(p):
         return {"emotion": ears.emotion_read(p),
                 "features": feats.features(p, rhythm=True)["features"],
                 "sample": feats.sample_profile(p)}
-    cid, result = _cached("analyze", params, data, file.filename, _compute, response)
-    return {"filename": file.filename, "content_id": cid, **result}
+    cid, result, source = await _run(file, "analyze", params, _compute, response)
+    return {"filename": file.filename, "content_id": cid, **result, "source": source}
 
 
 @v1.post("/grade", operation_id="grade")
@@ -388,22 +430,22 @@ async def grade(response: Response, file: UploadFile = File(...), role: str = ""
     0..1 + S/A/B/C/D tier, per-rule sub-scores (seam click, zero-crossing
     cleanliness, DC, bar self-consistency, level, bass mono-compat) + flags.
     `role` ('bass'|'drums'|…) is a hint. WAV/FLAC preferred. Torch-free, cached."""
-    data = await _read_upload(file)
-    params = {"role": role or None, "v": 1}
-    cid, result = _cached("grade", params, data, file.filename,
-                          lambda p: grade_eng.grade(p, role=role or None).model_dump(), response)
-    return {"filename": file.filename, "content_id": cid, "grade": result}
+    cid, result, source = await _run(file, "grade", {"role": role or None, "v": 1},
+                                     lambda p: grade_eng.grade(p, role=role or None).model_dump(),
+                                     response)
+    return {"filename": file.filename, "content_id": cid, "grade": result, "source": source}
 
 
 @v1.post("/loudness", operation_id="loudness")
 async def loudness(response: Response, file: UploadFile = File(...),
                    _p: auth.Principal = Depends(require)):
     """Upload audio → BS.1770 integrated LUFS + sample/true peak + crest + the
-    gain (dB) to hit each delivery target (-14 streaming, -9 club). Torch-free, cached."""
-    data = await _read_upload(file)
-    cid, result = _cached("loudness", {"v": 1}, data, file.filename,
+    gain (dB) to hit each delivery target (-14 streaming, -9 club). Torch-free, cached.
+    Full-track: peak/true-peak/RMS are exact; integrated LUFS is chunked-approximate
+    (`approximate:true`) — pyloudnorm gating can't merge chunks exactly."""
+    cid, result, source = await _run(file, "loudness", {"v": 2},
                                      loudness_eng.loudness, response)
-    return {"filename": file.filename, "content_id": cid, **result}
+    return {"filename": file.filename, "content_id": cid, **result, "source": source}
 
 
 @v1.post("/spectrum", operation_id="spectrum")
@@ -412,22 +454,21 @@ async def spectrum(response: Response, file: UploadFile = File(...),
                    _p: auth.Principal = Depends(require)):
     """FFT-as-a-service: a downsampled, render-ready spectrogram (`bands`×`frames`,
     0..1). `mel=` perceptual vs log-linear; `frames=1` → a single averaged FFT
-    spectrum. Bounded payload → caches cheaply. Torch-free."""
-    data = await _read_upload(file)
-    params = {"bands": bands, "frames": frames, "mel": mel, "v": 1}
-    cid, result = _cached("spectrum", params, data, file.filename,
+    spectrum. Bounded payload → caches cheaply. Torch-free. Full-track: frames span
+    the WHOLE track (streamed windows binned into the global time grid)."""
+    params = {"bands": bands, "frames": frames, "mel": mel, "v": 2}
+    cid, result, source = await _run(file, "spectrum", params,
                                      lambda p: signal.spectrum(p, bands=bands, frames=frames, mel=mel),
                                      response)
-    return {"filename": file.filename, "content_id": cid, **result}
+    return {"filename": file.filename, "content_id": cid, **result, "source": source}
 
 
 @v1.post("/naming", operation_id="naming")
 async def naming(response: Response, file: UploadFile = File(...), index: int = 1,
                  _p: auth.Principal = Depends(require)):
     """Upload a sample → a convention-compliant name (NN_role_character) derived
-    from the MEASURED role + character, not the file name. Torch-free, cached."""
-    data = await _read_upload(file)
-
+    from the MEASURED role + character, not the file name. Torch-free, cached.
+    Clip-tier: first ~30 s."""
     def _compute(p):
         prof = feats.sample_profile(p)
         ff = feats.features(p, rhythm=False)["features"]
@@ -435,8 +476,8 @@ async def naming(response: Response, file: UploadFile = File(...), index: int =
         suggested = naming_eng.suggest_name(index, prof["role"], character=character)
         return {"suggested_name": suggested, "role": prof["role"], "character": character,
                 "stem": naming_eng.stem_of(prof["role"]), "lint": naming_eng.lint(suggested)}
-    cid, result = _cached("naming", {"index": index, "v": 1}, data, file.filename, _compute, response)
-    return {"filename": file.filename, "content_id": cid, **result}
+    cid, result, source = await _run(file, "naming", {"index": index, "v": 1}, _compute, response)
+    return {"filename": file.filename, "content_id": cid, **result, "source": source}
 
 
 # ── #37 separation seam — no CPU path on erable; env-selected GPU backend later ──

armada/api/cache.py

@@ -21,18 +21,27 @@ import db
 # ── addressing ──────────────────────────────────────────────────────────────
 def content_id(audio_path: str | Path) -> str:
     """sha256 of decoded PCM (+ samplerate). Falls back to raw file bytes if the
-    format can't be decoded here (still dedupes identical files)."""
+    format can't be decoded here (still dedupes identical files).
+
+    STREAMED (#7): hashes the decoded PCM block-by-block instead of reading the
+    whole file into RAM — the concatenated float32 bytes are identical to the old
+    whole-array hash, so existing cache ids stay valid. Bounded memory either way."""
     try:
         import numpy as np
         import soundfile as sf
-        y, sr = sf.read(str(audio_path), dtype="float32", always_2d=False)
         h = hashlib.sha256()
-        h.update(np.ascontiguousarray(y).tobytes())
+        with sf.SoundFile(str(audio_path)) as f:
+            sr = f.samplerate
+            for block in f.blocks(blocksize=1 << 18, dtype="float32", always_2d=False):
+                h.update(np.ascontiguousarray(block).tobytes())
         h.update(str(sr).encode())
         return "cid_" + h.hexdigest()[:32]
     except Exception:
-        raw = Path(audio_path).read_bytes()
-        return "raw_" + hashlib.sha256(raw).hexdigest()[:32]
+        h = hashlib.sha256()
+        with open(audio_path, "rb") as fp:
+            for chunk in iter(lambda: fp.read(1 << 20), b""):
+                h.update(chunk)
+        return "raw_" + h.hexdigest()[:32]
 
 
 def params_hash(params: dict | None) -> str:

armada/api/config.py

@@ -42,7 +42,11 @@ DB_PATH = Path(os.environ.get("FOURIER_DB", HERE / "_cache" / "fourier.db"))
 DEV_TOKEN = os.environ.get("FOURIER_DEV_TOKEN", "")
 
 # ── limits ─────────────────────────────────────────────────────────────────────
-MAX_UPLOAD_MB = int(os.environ.get("FOURIER_MAX_UPLOAD_MB", "40"))
+# Match the gateway's client_max_body_size (100 MB) so the app never rejects a
+# body nginx let through. Safe to accept large uploads now: ingest + hashing +
+# the full-track engines all stream in bounded windows (#7), so peak memory is
+# O(window), not O(file). The gateway's friendly 413 fires first above this.
+MAX_UPLOAD_MB = int(os.environ.get("FOURIER_MAX_UPLOAD_MB", "100"))
 
 # ── artifact store (#27) — local disk on erable; freebox deferred ───────────────
 # Big binaries (fetched sources, stems, loops) live content-addressed under here,

armada/api/engines/loudness.py

@@ -5,11 +5,20 @@ The mastering numbers PLN + hexa actually gate on (reference_postprod_master:
 integrated loudness; true-peak is a cheap 4× oversample (catches inter-sample
 peaks a raw sample-peak misses). Returns gain-to-target so a caller knows exactly
 how much to push for each delivery target.
+
+FULL-TRACK (#7): streams the WHOLE track in bounded windows. Peak / true-peak /
+RMS / crest are EXACT (running max / accumulated sum-of-squares). Integrated LUFS
+is energy-weighted across per-window meters — pyloudnorm's gating can't be merged
+exactly across chunks, so this is a close APPROXIMATION, flagged `approximate:true`
+with `lufs_method:"chunked"`. For a steady master the error is well under ~0.5 LU.
 """
 import numpy as np
 
+from . import streaming
+
 _EPS = 1e-12
 TARGETS = {"streaming": -14.0, "club": -9.0}   # LUFS, reference_postprod_master
+_METER_MIN_S = 0.4                              # pyloudnorm needs ≥ one 400 ms gated block
 
 
 def _db(x):
@@ -26,37 +35,65 @@ def _true_peak_dbfs(mono, sr):
         return _db(np.max(np.abs(mono)))
 
 
-def loudness(audio_path, max_seconds=180.0) -> dict:
-    """Integrated LUFS, sample/true peak (dBFS), crest, + gain-to-target."""
-    import soundfile as sf
-    data, sr = sf.read(str(audio_path), dtype="float64")
-    if data.shape[0] < sr // 2:          # pyloudnorm needs ≥ ~0.4 s for a gated block
+def loudness(audio_path, win_s=streaming.DEFAULT_WIN_S) -> dict:
+    """Integrated LUFS (chunked-approx), sample/true peak (dBFS, exact), crest,
+    + gain-to-target, over the whole track."""
+    meta = streaming.probe(audio_path)
+    sr, channels = meta["sr"], meta["channels"]
+    if meta["frames"] and meta["frames"] < sr // 2:
         raise ValueError("clip too short for integrated loudness (need ≥ 0.5 s)")
-    if max_seconds:
-        data = data[: int(sr * max_seconds)]
-    mono = data if data.ndim == 1 else data.mean(axis=1)
 
-    rms_db = _db(np.sqrt(np.mean(mono ** 2) + _EPS))
-    peak_db = _db(np.max(np.abs(mono)))
-    tp_db = _true_peak_dbfs(mono, sr)
-
-    lufs = None
     try:
         import pyloudnorm as pyln
-        lufs = float(pyln.Meter(sr).integrated_loudness(data))
+        meter = pyln.Meter(sr)
+    except Exception:
+        meter = None
+
+    sumsq = 0.0
+    n = 0
+    peak = _EPS
+    tp = _EPS
+    energy_sum = 0.0        # Σ dur_i · 10^(LUFS_i/10) — linear loudness, duration-weighted
+    dur_sum = 0.0
+    for block, sr, _ in streaming.windows(audio_path, win_s, mono=False, dtype="float64"):
+        if not len(block):
+            continue
+        mono = block if block.ndim == 1 else block.mean(axis=1)
+        sumsq += float(np.sum(mono ** 2))
+        n += mono.size
+        peak = max(peak, float(np.max(np.abs(mono))))
+        tp = max(tp, 10 ** (_true_peak_dbfs(mono, sr) / 20.0))
+        if meter is not None and len(block) >= int(_METER_MIN_S * sr):
+            try:
+                lw = float(meter.integrated_loudness(block))
             except Exception:
-        pass
+                lw = float("-inf")
+            if np.isfinite(lw):                        # silence-gated windows → skip
+                w = len(block) / sr
+                energy_sum += w * (10 ** (lw / 10.0))
+                dur_sum += w
+
+    if n < sr // 2:
+        raise ValueError("clip too short for integrated loudness (need ≥ 0.5 s)")
 
-    gain = ({k: round(t - lufs, 2) for k, t in TARGETS.items()} if lufs is not None
-            and np.isfinite(lufs) else None)
+    rms_db = _db(np.sqrt(sumsq / max(n, 1) + _EPS))
+    peak_db = _db(peak)
+    tp_db = _db(tp)
+    lufs = 10.0 * np.log10(energy_sum / dur_sum) if dur_sum > 0 else None
+    gain = ({k: round(t - lufs, 2) for k, t in TARGETS.items()}
+            if lufs is not None and np.isfinite(lufs) else None)
     return {
         "lufs_integrated": round(lufs, 2) if lufs is not None and np.isfinite(lufs) else None,
+        "lufs_method": "chunked",
+        "approximate": True,          # energy-weighted across windows, not a single gated pass
         "peak_dbfs": round(peak_db, 2),
         "true_peak_dbfs": round(tp_db, 2),
         "rms_dbfs": round(rms_db, 2),
         "crest_db": round(peak_db - rms_db, 2),
         "gain_to_target_db": gain,    # add this many dB to hit each LUFS target
-        "channels": 1 if data.ndim == 1 else data.shape[1],
+        "channels": int(channels),
         "sr": int(sr),
+        "duration_s": round(meta["duration_s"], 3),
         "engine": "light",
+        "full_track": True,
     }

armada/api/engines/signal.py

-"""Light signal building blocks — onsets + waveform peaks. CPU, NO torch.
+"""Light signal building blocks — onsets + waveform + spectrum. CPU, NO torch.
 
 The cheapest analysis tier: thin librosa wrappers that need no model at all.
 Especially useful for hexa (a visuals platform): /waveform gives a render-ready
 peak/energy envelope, /onsets gives rhythmic hit times for visual sync + slicing.
-Self-contained like ears_light/feats (the container has only armada/api/).
+
+FULL-TRACK (#7): these stream the WHOLE upload in bounded windows and aggregate,
+instead of only the first 60 s — so a long track is analyzed end-to-end without
+buffering it all in RAM. waveform/spectrum aggregate EXACTLY (fixed output grid
+filled by streaming reduction); onsets builds one full-track onset envelope (tiny
+— hop-downsampled) from per-window pieces. Native sample rate throughout (no
+per-block resample, which would seam at window edges). Self-contained like
+ears_light/feats (the container has only armada/api/).
 """
 import numpy as np
 
-SR = 22050
-MAX_S = 60.0
+from . import streaming
+
+HOP = 512                 # librosa default; fixes frame→time mapping across windows
+_MIN_FRAMES_SR = 20       # need ≥ 1/20 s of audio to say anything
+_FALLBACK_CAP_S = 600.0   # bounded whole-file decode when a format isn't frame-countable
 
 
-def _load(audio_path, max_seconds):
+def _load_capped(audio_path, sr_target=22050):
+    """Bounded whole-file fallback for formats soundfile can't frame-count.
+    Decodes at most _FALLBACK_CAP_S seconds — never the unbounded file."""
     import librosa
-    y, _ = librosa.load(str(audio_path), sr=SR, mono=True, duration=max_seconds)
-    if y.size < SR // 20:
+    y, sr = librosa.load(str(audio_path), sr=sr_target, mono=True, duration=_FALLBACK_CAP_S)
+    if y.size < sr // _MIN_FRAMES_SR:
         raise ValueError("clip too short / undecodable (need ≥ 50 ms of audio)")
-    return y
+    return y, sr
 
 
-def onsets(audio_path, max_seconds=MAX_S) -> dict:
-    """Onset hit times (s) + tempo + onset rate. Drift-tolerant tempo via the
-    onset envelope (librosa 0.11 feature.tempo, not the removed rhythm.tempo)."""
+def onsets(audio_path, win_s=streaming.DEFAULT_WIN_S) -> dict:
+    """Onset hit times (s) + tempo + onset rate over the WHOLE track. Builds one
+    full-track onset-strength envelope by concatenating per-window pieces (the
+    envelope is hop-downsampled, so a 10-min track is only ~tens of KB), then runs
+    detection + drift-tolerant tempo on it once."""
     import librosa
-    y = _load(audio_path, max_seconds)
-    onset_env = librosa.onset.onset_strength(y=y, sr=SR)
-    times = librosa.onset.onset_detect(onset_envelope=onset_env, sr=SR, units="time")
-    tempo = float(np.atleast_1d(librosa.feature.tempo(onset_envelope=onset_env, sr=SR))[0])
-    dur = len(y) / SR
+    meta = streaming.probe(audio_path)
+    sr = meta["sr"]
+    if meta["frames"] <= 0:                       # uncountable format → bounded fallback
+        y, sr = _load_capped(audio_path)
+        env = librosa.onset.onset_strength(y=y, sr=sr, hop_length=HOP)
+        full = False
+    else:
+        pieces, full = [], True
+        for block, sr, _ in streaming.windows(audio_path, win_s, mono=True):
+            if block.size:
+                pieces.append(librosa.onset.onset_strength(y=block, sr=sr, hop_length=HOP))
+        if not pieces:
+            raise ValueError("clip too short / undecodable (need ≥ 50 ms of audio)")
+        env = np.concatenate(pieces)
+
+    times = librosa.onset.onset_detect(onset_envelope=env, sr=sr, hop_length=HOP, units="time")
+    tempo = float(np.atleast_1d(librosa.feature.tempo(onset_envelope=env, sr=sr, hop_length=HOP))[0])
+    dur = meta["duration_s"] or (len(env) * HOP / sr)
     return {
         "tempo_bpm": round(tempo, 2),
         "onsets": [round(float(t), 4) for t in times],
         "n_onsets": int(len(times)),
         "onset_rate": round(len(times) / dur, 3) if dur else 0.0,
         "duration_s": round(dur, 3),
-        "sr": SR,
+        "sr": int(sr),
         "engine": "light",
+        "full_track": full,
     }
 
 
-def waveform(audio_path, bins=800, max_seconds=MAX_S) -> dict:
-    """Downsampled waveform for rendering: per-bin [min, max] in [-1,1] (draws the
-    classic waveform) + a 0..1 RMS energy envelope (audio-reactive intensity)."""
-    y = _load(audio_path, max_seconds)
+def waveform(audio_path, bins=800, win_s=streaming.DEFAULT_WIN_S) -> dict:
+    """Downsampled waveform for rendering: per-bin [min, max] in [-1,1] + a 0..1
+    RMS energy envelope, spanning the WHOLE track. Exact: each `bins` bucket is
+    filled by an unsegmented min/max/mean-square reduction over the streamed
+    samples, so the result equals a single-pass computation."""
     bins = max(1, min(int(bins), 4000))
+    meta = streaming.probe(audio_path)
+    total, sr = meta["frames"], meta["sr"]
+
+    if total <= 0:                                 # uncountable format → bounded fallback
+        y, sr = _load_capped(audio_path)
+        return _waveform_array(y, sr, bins, full=False)
+
+    mn = np.full(bins, np.inf, dtype="float64")
+    mx = np.full(bins, -np.inf, dtype="float64")
+    sq = np.zeros(bins, dtype="float64")
+    cnt = np.zeros(bins, dtype="int64")
+    for block, sr, start in streaming.windows(audio_path, win_s, mono=True):
+        if not block.size:
+            continue
+        idx = ((np.arange(len(block), dtype="int64") + start) * bins // total)
+        np.clip(idx, 0, bins - 1, out=idx)
+        np.minimum.at(mn, idx, block)
+        np.maximum.at(mx, idx, block)
+        np.add.at(sq, idx, block.astype("float64") ** 2)
+        np.add.at(cnt, idx, 1)
+    if cnt.sum() < sr // _MIN_FRAMES_SR:
+        raise ValueError("clip too short / undecodable (need ≥ 50 ms of audio)")
+    return _finish_waveform(mn, mx, sq, cnt, bins, total, sr, full=True)
+
+
+def _waveform_array(y, sr, bins, full):
     edges = np.linspace(0, len(y), bins + 1).astype(int)
-    peaks, rms = [], []
+    mn = np.zeros(bins); mx = np.zeros(bins); sq = np.zeros(bins); cnt = np.zeros(bins, dtype="int64")
     for i in range(bins):
         seg = y[edges[i]:edges[i + 1]]
-        if seg.size == 0:
-            peaks.append([0.0, 0.0]); rms.append(0.0); continue
-        peaks.append([round(float(seg.min()), 4), round(float(seg.max()), 4)])
-        rms.append(float(np.sqrt(np.mean(seg ** 2))))
-    norm = max(rms) or 1.0
+        if seg.size:
+            mn[i], mx[i] = seg.min(), seg.max()
+            sq[i] = np.sum(seg.astype("float64") ** 2); cnt[i] = seg.size
+    return _finish_waveform(mn, mx, sq, cnt, bins, len(y), sr, full=full)
+
+
+def _finish_waveform(mn, mx, sq, cnt, bins, total, sr, full):
+    empty = cnt == 0
+    mn = np.where(empty, 0.0, mn)
+    mx = np.where(empty, 0.0, mx)
+    rms = np.sqrt(np.where(cnt > 0, sq / np.maximum(cnt, 1), 0.0))
+    norm = float(rms.max()) or 1.0
     return {
         "bins": bins,
-        "peaks": peaks,
-        "rms": [round(r / norm, 4) for r in rms],   # 0..1, peak-normalized
-        "duration_s": round(len(y) / SR, 3),
-        "sr": SR,
+        "peaks": [[round(float(lo), 4), round(float(hi), 4)] for lo, hi in zip(mn, mx)],
+        "rms": [round(float(r / norm), 4) for r in rms],
+        "duration_s": round(total / sr, 3) if sr else 0.0,
+        "sr": int(sr),
         "engine": "light",
+        "full_track": full,
     }
 
 
-def spectrum(audio_path, bands=64, frames=200, mel=True, max_seconds=MAX_S) -> dict:
+def spectrum(audio_path, bands=64, frames=200, mel=True, win_s=streaming.DEFAULT_WIN_S) -> dict:
     """FFT-as-a-service: a downsampled, render-ready spectrogram — `bands` ×
-    `frames`, 0..1 normalized (from dB, ref=peak). mel=True uses perceptual mel
-    bands (good for visuals); mel=False uses log-spaced linear bins. `frames=1`
-    collapses to a single time-averaged FFT magnitude spectrum. Bounded payload,
-    so it caches cheaply. Torch-free."""
+    `frames`, 0..1 normalized, spanning the WHOLE track. Streams windows, bins each
+    window's spectrogram columns into the global `frames` grid by absolute time,
+    accumulates mean POWER per (band, frame), then converts to dB (ref = the
+    track's global peak power) once at the end. `frames=1` → a single time-averaged
+    spectrum. Torch-free."""
     import librosa
-    y = _load(audio_path, max_seconds)
     bands = max(1, min(int(bands), 256))
     frames = max(1, min(int(frames), 1000))
+    meta = streaming.probe(audio_path)
+    total_frames, sr = meta["frames"], meta["sr"]
+
+    if total_frames <= 0:                          # uncountable format → bounded fallback
+        y, sr = _load_capped(audio_path)
+        total_frames = len(y)
+        blocks = [(y, sr, 0)]
+        full = False
+    else:
+        blocks = streaming.windows(audio_path, win_s, mono=True)
+        full = True
+
+    acc = np.zeros((bands, frames), dtype="float64")   # summed power per (band, global-frame)
+    cnt = np.zeros(frames, dtype="int64")
+    gmax = 1e-12                                        # global peak power for the dB ref
+    band_hz = None
+    total_dur = (total_frames / sr) if sr else 0.0
+
+    for block, sr, start in blocks:
+        if block.size < HOP:
+            continue
         if mel:
-        S = librosa.feature.melspectrogram(y=y, sr=SR, n_mels=bands)
-        band_hz = [round(float(f), 1) for f in librosa.mel_frequencies(n_mels=bands, fmax=SR / 2)]
+            S = librosa.feature.melspectrogram(y=block, sr=sr, n_mels=bands, hop_length=HOP)
+            if band_hz is None:
+                band_hz = [round(float(f), 1) for f in librosa.mel_frequencies(n_mels=bands, fmax=sr / 2)]
         else:
-        mag = np.abs(librosa.stft(y, n_fft=2048, hop_length=512)) ** 2
-        freqs = librosa.fft_frequencies(sr=SR, n_fft=2048)
-        edges = np.logspace(np.log10(20), np.log10(SR / 2), bands + 1)
+            mag = np.abs(librosa.stft(block, n_fft=2048, hop_length=HOP)) ** 2
+            freqs = librosa.fft_frequencies(sr=sr, n_fft=2048)
+            edges = np.logspace(np.log10(20), np.log10(sr / 2), bands + 1)
             S = np.stack([mag[(freqs >= edges[i]) & (freqs < edges[i + 1])].sum(axis=0)
                           for i in range(bands)])
+            if band_hz is None:
                 band_hz = [round(float(np.sqrt(edges[i] * edges[i + 1])), 1) for i in range(bands)]
-    Sdb = librosa.power_to_db(S + 1e-12, ref=np.max)          # [-80..0] dB
-    # downsample time → `frames` columns by averaging within each block
-    cols = Sdb.shape[1]
-    edges = np.linspace(0, cols, frames + 1).astype(int)
-    blocks = [Sdb[:, edges[i]:edges[i + 1]].mean(axis=1) if edges[i + 1] > edges[i]
-              else Sdb[:, min(edges[i], cols - 1)] for i in range(frames)]
-    spec = (np.clip(np.stack(blocks, axis=1), -80, 0) + 80) / 80.0   # bands×frames, 0..1
+        gmax = max(gmax, float(S.max()))
+        # map each column to its absolute time → global frame bucket
+        col_t = (start + np.arange(S.shape[1]) * HOP) / sr
+        gf = np.zeros(S.shape[1], dtype="int64") if total_dur <= 0 else \
+            np.clip((col_t / total_dur * frames).astype("int64"), 0, frames - 1)
+        for b in range(frames):
+            sel = gf == b
+            if sel.any():
+                acc[:, b] += S[:, sel].sum(axis=1)
+                cnt[b] += int(sel.sum())
+
+    if cnt.sum() == 0:
+        raise ValueError("clip too short / undecodable (need ≥ 50 ms of audio)")
+    # empty frame buckets (sparse coverage) borrow the nearest filled one's mean
+    mean_p = np.zeros((bands, frames), dtype="float64")
+    filled = cnt > 0
+    mean_p[:, filled] = acc[:, filled] / cnt[filled]
+    if not filled.all():
+        idx = np.where(filled)[0]
+        for b in np.where(~filled)[0]:
+            mean_p[:, b] = mean_p[:, idx[np.argmin(np.abs(idx - b))]]
+    Sdb = librosa.power_to_db(mean_p + 1e-12, ref=gmax)
+    spec = (np.clip(Sdb, -80, 0) + 80) / 80.0
     return {
         "bands": bands,
         "frames": frames,
         "mel": bool(mel),
-        "band_hz": band_hz,
-        "spec": [[round(float(v), 4) for v in row] for row in spec],   # [band][frame]
-        "duration_s": round(len(y) / SR, 3),
-        "sr": SR,
+        "band_hz": band_hz or [],
+        "spec": [[round(float(v), 4) for v in row] for row in spec],
+        "duration_s": round(total_dur, 3),
+        "sr": int(sr),
         "engine": "light",
+        "full_track": full,
     }

armada/api/engines/streaming.py

+"""Bounded-memory streaming primitives for full-track analysis. CPU, NO torch.
+
+Producers upload whole tracks, not 60 s clips. Rather than 413-reject a big file
+(or buffer the whole thing in RAM), the signal/loudness engines decode it in
+fixed time windows and aggregate as they go — so peak memory is O(one window),
+not O(file), and a large upload can't OOM the shared erable box.
+
+soundfile (libsndfile 1.2) streams WAV/FLAC/OGG/MP3 with exact frame counts, so
+we read at NATIVE sample rate and let each engine pass `sr` to librosa — no
+per-block resampling (which would add seam artifacts at window boundaries).
+"""
+import numpy as np
+
+# 30 s windows: at 96 kHz stereo float64 that is ~44 MB resident — comfortably
+# under the container's 3 GB cap, with margin for the engine's own working set.
+DEFAULT_WIN_S = 30.0
+
+
+def probe(path) -> dict:
+    """Header-only metadata (no decode): frames, sr, channels, duration_s.
+    `frames <= 0` means the container can't frame-count this file (rare); callers
+    fall back to a bounded whole-file decode."""
+    import soundfile as sf
+    i = sf.info(str(path))
+    return {
+        "frames": int(i.frames),
+        "sr": int(i.samplerate),
+        "channels": int(i.channels),
+        "duration_s": (i.frames / i.samplerate) if i.samplerate else 0.0,
+        "format": i.format,
+    }
+
+
+def windows(path, win_s=DEFAULT_WIN_S, *, mono=True, dtype="float32"):
+    """Yield (block, sr, start_frame) over the WHOLE file in `win_s` windows.
+
+    `mono=True` downmixes channels (mean); `mono=False` yields 2-D (n, channels).
+    Non-overlapping and contiguous: start_frame of window k is the running sum of
+    prior window lengths, so callers can map a block's samples to absolute time.
+    Bounded memory: only one window is resident at a time."""
+    import soundfile as sf
+    with sf.SoundFile(str(path)) as f:
+        sr = f.samplerate
+        wf = max(1, int(win_s * sr))
+        start = 0
+        while True:
+            blk = f.read(frames=wf, dtype=dtype, always_2d=not mono)
+            if len(blk) == 0:
+                return
+            block = (blk if blk.ndim == 1 else blk.mean(axis=1)) if mono else blk
+            yield block, sr, start
+            start += len(block)

armada/api/tests/test_fulltrack.py

+"""Full-track streaming (#7) — the guarantees that make big uploads safe.
+
+Proves the two things unit tests on 2 s clips can't: (1) the signal/loudness
+engines now analyse the WHOLE track, past the old 60 s / 180 s head cap, and
+(2) streaming didn't change the numbers — content_id stays byte-identical to the
+old whole-file hash (so the deployed cache stays warm), and exact reductions
+(waveform min/max) equal a single-pass computation.
+
+    cd armada/api && python -m pytest tests/test_fulltrack.py -q
+"""
+import hashlib
+import tempfile
+from pathlib import Path
+
+import numpy as np
+import soundfile as sf
+
+import cache
+from engines import loudness, signal, streaming
+
+SR = 22050
+
+
+def _click_track(bpm=120, secs=70.0, sr=SR):
+    """A long pulse train (>60 s) so a head-only analysis would visibly truncate."""
+    n = int(sr * secs)
+    y = np.zeros(n, dtype="float32")
+    step = int(sr * 60 / bpm)
+    for i in range(0, n, step):
+        y[i:i + 200] += np.hanning(min(200, n - i)).astype("float32")
+    return y
+
+
+def _write(y, sr=SR, suffix=".wav"):
+    p = Path(tempfile.mktemp(suffix=suffix))
+    sf.write(str(p), y, sr)
+    return str(p)
+
+
+# ── (1) full-track coverage: past the old 60 s / 180 s head cap ─────────────────
+def test_onsets_cover_whole_track_not_just_head():
+    p = _write(_click_track(secs=70.0))
+    r = signal.onsets(p)
+    assert r["full_track"] is True
+    assert r["duration_s"] >= 69.0                 # whole track, not capped at 60
+    assert max(r["onsets"]) > 60.0                 # hits found beyond the old head window
+    assert r["n_onsets"] >= 120                     # ~120 bpm × 70 s ≈ 140 pulses
+
+
+def test_waveform_spans_whole_track():
+    p = _write(_click_track(secs=70.0))
+    r = signal.waveform(p, bins=100)
+    assert r["full_track"] is True and r["bins"] == 100
+    assert r["duration_s"] >= 69.0
+    # pulses recur throughout → energy present in the last bins, not only the first
+    assert r["rms"][-1] > 0.0 and r["rms"][90] > 0.0
+
+
+def test_loudness_full_track_exact_peak_and_flagged_approx():
+    y = _click_track(secs=70.0) * 0.5
+    r = loudness.loudness(_write(y))
+    assert r["full_track"] is True and r["approximate"] is True
+    assert r["lufs_method"] == "chunked" and r["lufs_integrated"] is not None
+    # peak is an EXACT running max over the stream
+    assert abs(r["peak_dbfs"] - 20 * np.log10(float(np.max(np.abs(y))))) < 0.1
+    assert r["duration_s"] >= 69.0
+
+
+# ── (2) streaming didn't change the numbers ─────────────────────────────────────
+def test_content_id_streaming_equals_whole_file_hash():
+    """The streamed content_id must equal the OLD whole-array hash, or every
+    deployed cache entry silently invalidates on deploy."""
+    y = (0.2 * np.sin(2 * np.pi * 220 * np.linspace(0, 5, 5 * SR, endpoint=False))).astype("float32")
+    p = _write(y)
+    # reproduce the pre-#7 computation: sha256(whole_pcm.tobytes() + str(sr))
+    data, sr = sf.read(p, dtype="float32", always_2d=False)
+    h = hashlib.sha256()
+    h.update(np.ascontiguousarray(data).tobytes())
+    h.update(str(sr).encode())
+    expected = "cid_" + h.hexdigest()[:32]
+    assert cache.content_id(p) == expected
+
+
+def test_waveform_reduction_is_exact_vs_single_pass():
+    """Streamed per-bin min/max must equal a single-pass numpy computation."""
+    y = (0.4 * np.sin(2 * np.pi * 110 * np.linspace(0, 8, 8 * SR, endpoint=False))).astype("float32")
+    p = _write(y)
+    bins = 64
+    r = signal.waveform(p, bins=bins)
+    edges = np.linspace(0, len(y), bins + 1).astype(int)
+    for i in range(bins):
+        seg = y[edges[i]:edges[i + 1]]
+        assert abs(r["peaks"][i][0] - round(float(seg.min()), 4)) < 1e-3
+        assert abs(r["peaks"][i][1] - round(float(seg.max()), 4)) < 1e-3
+
+
+def test_streaming_windows_cover_all_samples_once():
+    """The window iterator must yield every sample exactly once, contiguously."""
+    y = np.arange(int(2.5 * SR), dtype="float32")
+    p = _write(y)
+    seen = 0
+    last_end = 0
+    for block, sr, start in streaming.windows(p, win_s=1.0, mono=True):
+        assert start == last_end                    # contiguous, no gap/overlap
+        seen += len(block)
+        last_end = start + len(block)
+    assert seen == len(y)