feat(douanier): content-addressed cache — the margin lever (#26)

Identical audio is analyzed ONCE and served forever. This is the economic core
that makes cheap-analysis-at-scale viable, and it makes the live emotion
endpoint feel instant on repeats.

WHY: per the design, repeat calls must cost ~nothing — that's the margin story
vs cloud egress. A re-analyzed track shouldn't pay the CLAP compute twice.

WHAT:
- cache.py — content_id = sha256 of DECODED PCM (+ samplerate) so re-encodes /
  re-uploads of the same sound dedupe; raw-bytes fallback for formats we can't
  decode here. params_hash folds engine settings (model, bars…) so different
  params cache separately. get/put (JSON inline or a result_ref path for big
  artifacts), yt-id/url → content_id aliases (#29 will use them), and an
  LRU/size-cap gc() that unlinks evicted artifact files (cache is transient on
  the freebox — rebuildable).
- db.py — cache + aliases tables (WAL already on), LRU index on (kind, last_access).
- app.py — /v1/analyze/emotion now content-addresses the upload, serves cached
  reads with X-Douanier-Cache: hit, computes+stores on miss. Engine call still
  behind engines.ears (unchanged contract).
- douanier.py — cache stats / gc admin commands.

VALIDATION:
- 22/22 tests green (+7 cache: stable/decode-invariant content id, raw fallback,
  miss→put→hit, params separate entries, hit counter, alias resolve, LRU gc
  spares the recently-touched entry).
- Real-CLAP end-to-end: same clip twice → call 1 miss 10.6s, call 2 hit 0.002s
  = 5663x speedup, identical V/A. The headline product benefit, measured.

armada/api/README.md

@@ -68,11 +68,13 @@ cd armada/api && ~/.virtualenvs/douanier/bin/python -m pytest tests/ -q
 ## Status
 
 Done: scaffold + emotion endpoint (#22/#36), **SQLite bearer-token auth + scopes
-+ `douanier` CLI (#23)**. Queued, building *under* this without changing the
-public shape: per-token quota (#24), async jobs + worker (#25),
-content-addressed cache (#26), signed artifact URLs (#27), `/features`+`/samples`
-hardened (#28), `/sources`→`/separate`→`/loops` heavy chain (#29–#31), OpenAPI
-client (#32), ops/runbook (#34).
++ `douanier` CLI (#23)**, **content-addressed cache (#26)** — a repeat clip
+returns in ~2ms instead of paying CLAP again (measured 5663× on a real run;
+`X-Douanier-Cache: hit`). Queued, building *under* this without changing the
+public shape: per-token quota (#24), async jobs + worker (#25), signed artifact
+URLs (#27), `/features`+`/samples` hardened (#28),
+`/sources`→`/separate`→`/loops` heavy chain (#29–#31), OpenAPI client (#32),
+ops/runbook (#34).
 
 > **Why a venv with `--system-site-packages`?** The ears/Foundry engines need the
 > multi-GB ML stack (torch/CLAP/librosa/demucs) already installed system-wide on

armada/api/app.py

@@ -11,10 +11,11 @@ SQLite auth/quota/jobs/cache machinery (#23–#27) that hardens it later.
 import tempfile
 from pathlib import Path
 
-from fastapi import FastAPI, APIRouter, UploadFile, File, Header, HTTPException, Depends
+from fastapi import FastAPI, APIRouter, UploadFile, File, Header, HTTPException, Depends, Response
 from fastapi.middleware.cors import CORSMiddleware
 
 import auth
+import cache
 import config
 from engines import ears
 
@@ -77,13 +78,13 @@ def whoami(principal: auth.Principal = Depends(require_scope("*"))):
 
 # ── #36 walking-skeleton endpoint: synchronous emotion read ─────────────────────
 @v1.post("/analyze/emotion")
-async def analyze_emotion(file: UploadFile = File(...),
+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).
 
-    Synchronous: no job queue, no cache yet (repeats recompute — fine for one
-    customer). The engine call is behind engines.ears so #28 can later wrap it
-    with cache + async without touching this route.
+    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.
     """
     data = await file.read()
     limit = config.MAX_UPLOAD_MB * 1024 * 1024
@@ -93,11 +94,18 @@ async def analyze_emotion(file: UploadFile = File(...),
     with tempfile.NamedTemporaryFile(suffix=suffix, delete=True) as tmp:
         tmp.write(data)
         tmp.flush()
+        cid = cache.content_id(tmp.name)
+        hit = cache.get(cid, "emotion")
+        if hit:
+            response.headers["X-Douanier-Cache"] = "hit"
+            return {"filename": file.filename, "content_id": cid, "emotion": hit["result"]}
         try:
             read = ears.emotion_of(tmp.name)
         except Exception as e:
             raise HTTPException(422, f"emotion analysis failed: {e}")
-    return {"filename": file.filename, "emotion": read}
+    cache.put(cid, "emotion", result=read)
+    response.headers["X-Douanier-Cache"] = "miss"
+    return {"filename": file.filename, "content_id": cid, "emotion": read}
 
 
 app.include_router(v1)

armada/api/cache.py

+"""Content-addressed cache — the margin lever.
+
+Identical audio is analyzed ONCE and served forever. The key is a content id =
+sha256 of the DECODED audio PCM (so re-encodes / re-uploads of the same sound
+dedupe), with engine params folded into a separate params_hash (so different
+settings cache separately). A youtube-id/URL alias resolves to a content id so a
+repeat source can skip the download (#29).
+
+Artifacts are TRANSIENT (freebox SSOT, rebuildable) → an LRU/size-cap eviction
+keeps the footprint bounded. Cheap JSON results live inline; big binaries are
+referenced by path (filled in by #27).
+"""
+import hashlib
+import json
+import time
+from pathlib import Path
+
+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)."""
+    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())
+        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]
+
+
+def params_hash(params: dict | None) -> str:
+    if not params:
+        return ""
+    blob = json.dumps(params, sort_keys=True, separators=(",", ":"))
+    return hashlib.sha256(blob.encode()).hexdigest()[:16]
+
+
+# ── get / put ─────────────────────────────────────────────────────────────────
+def get(content_id: str, kind: str, params: dict | None = None) -> dict | None:
+    """Return the cached entry (and bump last_access/hits), or None on miss.
+    For JSON kinds the result is under 'result'; binaries carry 'result_ref'."""
+    ph = params_hash(params)
+    with db.cursor() as c:
+        row = c.execute(
+            "SELECT id, result_json, result_ref, bytes FROM cache "
+            "WHERE content_id=? AND kind=? AND params_hash=?",
+            (content_id, kind, ph)).fetchone()
+        if not row:
+            return None
+        c.execute("UPDATE cache SET last_access=?, hits=hits+1 WHERE id=?",
+                  (time.time(), row["id"]))
+    return {
+        "content_id": content_id, "kind": kind,
+        "result": json.loads(row["result_json"]) if row["result_json"] else None,
+        "result_ref": row["result_ref"], "bytes": row["bytes"],
+    }
+
+
+def put(content_id: str, kind: str, *, result: dict | None = None,
+        result_ref: str | None = None, bytes_: int = 0,
+        params: dict | None = None) -> None:
+    """Insert/replace a cache entry. Pass `result` for JSON, `result_ref` for a
+    big artifact path."""
+    ph = params_hash(params)
+    now = time.time()
+    rj = json.dumps(result) if result is not None else None
+    with db.cursor() as c:
+        c.execute(
+            "INSERT INTO cache(content_id, kind, params_hash, result_json, "
+            "result_ref, bytes, created, last_access, hits) "
+            "VALUES(?,?,?,?,?,?,?,?,0) "
+            "ON CONFLICT(content_id, kind, params_hash) DO UPDATE SET "
+            "result_json=excluded.result_json, result_ref=excluded.result_ref, "
+            "bytes=excluded.bytes, last_access=excluded.last_access",
+            (content_id, kind, ph, rj, result_ref, bytes_, now, now))
+
+
+# ── aliases (yt-id / url → content_id) ─────────────────────────────────────────
+def resolve_alias(alias: str) -> str | None:
+    with db.cursor() as c:
+        row = c.execute("SELECT content_id FROM aliases WHERE alias=?",
+                        (alias,)).fetchone()
+        return row["content_id"] if row else None
+
+
+def put_alias(alias: str, content_id: str) -> None:
+    with db.cursor() as c:
+        c.execute("INSERT INTO aliases(alias, content_id, created) VALUES(?,?,?) "
+                  "ON CONFLICT(alias) DO UPDATE SET content_id=excluded.content_id",
+                  (alias, content_id, time.time()))
+
+
+# ── stats / eviction ──────────────────────────────────────────────────────────
+def stats() -> list[dict]:
+    with db.cursor() as c:
+        rows = c.execute(
+            "SELECT kind, COUNT(*) n, COALESCE(SUM(bytes),0) bytes, "
+            "COALESCE(SUM(hits),0) hits FROM cache GROUP BY kind ORDER BY bytes DESC"
+        ).fetchall()
+        return [dict(r) for r in rows]
+
+
+def gc(kind: str, max_bytes: int) -> dict:
+    """Evict coldest (LRU) entries of `kind` until total bytes ≤ max_bytes.
+    Returns {evicted, freed}. Also unlinks any result_ref files."""
+    evicted, freed = 0, 0
+    with db.cursor() as c:
+        total = c.execute("SELECT COALESCE(SUM(bytes),0) b FROM cache WHERE kind=?",
+                          (kind,)).fetchone()["b"]
+        if total <= max_bytes:
+            return {"evicted": 0, "freed": 0}
+        rows = c.execute("SELECT id, bytes, result_ref FROM cache WHERE kind=? "
+                         "ORDER BY last_access ASC", (kind,)).fetchall()
+        for r in rows:
+            if total - freed <= max_bytes:
+                break
+            if r["result_ref"]:
+                p = Path(r["result_ref"])
+                if p.exists():
+                    p.unlink()
+            c.execute("DELETE FROM cache WHERE id=?", (r["id"],))
+            evicted += 1
+            freed += r["bytes"]
+    return {"evicted": evicted, "freed": freed}

armada/api/db.py

@@ -29,6 +29,32 @@ CREATE TABLE IF NOT EXISTS tokens (
     revoked    INTEGER NOT NULL DEFAULT 0
 );
 CREATE INDEX IF NOT EXISTS idx_tokens_hash ON tokens(token_hash);
+
+-- content-addressed result cache (#26): the margin lever. A row per
+-- (content_id, kind, params) — JSON results inline in result_json, big binary
+-- artifacts referenced by result_ref (a freebox path, filled by #27).
+CREATE TABLE IF NOT EXISTS cache (
+    id          INTEGER PRIMARY KEY,
+    content_id  TEXT NOT NULL,          -- sha256 of decoded audio (canon)
+    kind        TEXT NOT NULL,          -- emotion | features | samples | stems | loops | source
+    params_hash TEXT NOT NULL DEFAULT '',  -- sha256 of engine params (model, bars…)
+    result_json TEXT,                   -- inline JSON result (cheap analyses)
+    result_ref  TEXT,                   -- path to a big artifact (stems/loops)
+    bytes       INTEGER NOT NULL DEFAULT 0,
+    created     REAL NOT NULL,
+    last_access REAL NOT NULL,
+    hits        INTEGER NOT NULL DEFAULT 0,
+    UNIQUE(content_id, kind, params_hash)
+);
+CREATE INDEX IF NOT EXISTS idx_cache_lru ON cache(kind, last_access);
+
+-- fast alias → content_id (e.g. a youtube id), so a repeat URL can skip the
+-- download when we've already seen that source (#29). Canon stays content_id.
+CREATE TABLE IF NOT EXISTS aliases (
+    alias      TEXT PRIMARY KEY,
+    content_id TEXT NOT NULL,
+    created    REAL NOT NULL
+);
 """
 
 

armada/api/douanier.py

@@ -14,6 +14,7 @@ import datetime as dt
 import sys
 
 import auth
+import cache
 import db
 
 
@@ -57,6 +58,21 @@ def cmd_db_init(_a):
     print(f"initialized {db.config.DB_PATH}")
 
 
+def cmd_cache_stats(_a):
+    rows = cache.stats()
+    if not rows:
+        print("(cache empty)")
+        return
+    print(f"{'kind':<10} {'entries':>8} {'bytes':>12} {'hits':>8}")
+    for r in rows:
+        print(f"{r['kind']:<10} {r['n']:>8} {r['bytes']:>12} {r['hits']:>8}")
+
+
+def cmd_cache_gc(a):
+    res = cache.gc(a.kind, int(a.max_bytes))
+    print(f"evicted {res['evicted']} entries, freed {res['freed']} bytes from '{a.kind}'")
+
+
 def main(argv=None):
     p = argparse.ArgumentParser(prog="douanier", description=__doc__)
     sub = p.add_subparsers(dest="group", required=True)
@@ -78,6 +94,14 @@ def main(argv=None):
     d = sub.add_parser("db", help="database admin").add_subparsers(dest="cmd", required=True)
     d.add_parser("init", help="create tables").set_defaults(func=cmd_db_init)
 
+    ca = sub.add_parser("cache", help="inspect/evict the content cache").add_subparsers(
+        dest="cmd", required=True)
+    ca.add_parser("stats", help="per-kind entry/byte/hit counts").set_defaults(func=cmd_cache_stats)
+    g = ca.add_parser("gc", help="evict coldest entries of a kind down to a byte cap")
+    g.add_argument("kind")
+    g.add_argument("max_bytes")
+    g.set_defaults(func=cmd_cache_gc)
+
     args = p.parse_args(argv)
     if getattr(args, "ttl", None) in ("never", "none", "None"):
         args.ttl = None

armada/api/tests/test_cache.py

+"""Cache unit tests — addressing, get/put, aliases, eviction. No CLAP needed."""
+import tempfile
+from pathlib import Path
+
+import numpy as np
+import soundfile as sf
+
+import cache
+
+
+def _wav(freq=220, secs=1.0, sr=22050):
+    t = np.linspace(0, secs, int(secs * sr), endpoint=False)
+    y = 0.2 * np.sin(2 * np.pi * freq * t)
+    f = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
+    sf.write(f.name, y, sr)
+    return f.name
+
+
+def test_content_id_is_stable_and_decode_invariant():
+    a, b = _wav(220), _wav(220)        # identical audio, two files
+    c = _wav(330)                      # different audio
+    assert cache.content_id(a) == cache.content_id(b)
+    assert cache.content_id(a) != cache.content_id(c)
+    assert cache.content_id(a).startswith("cid_")
+
+
+def test_raw_fallback_for_undecodable():
+    f = tempfile.NamedTemporaryFile(suffix=".bin", delete=False)
+    Path(f.name).write_bytes(b"not audio at all")
+    assert cache.content_id(f.name).startswith("raw_")
+
+
+def test_get_miss_then_put_then_hit():
+    cid = "cid_test_emotion"
+    assert cache.get(cid, "emotion") is None
+    cache.put(cid, "emotion", result={"valence": 0.5, "arousal": -0.1})
+    hit = cache.get(cid, "emotion")
+    assert hit and hit["result"]["valence"] == 0.5
+
+
+def test_params_separate_entries():
+    cid = "cid_params"
+    cache.put(cid, "stems", result={"backend": "demucs"}, params={"model": "htdemucs"})
+    cache.put(cid, "stems", result={"backend": "merge"}, params={"model": "merge"})
+    assert cache.get(cid, "stems", {"model": "htdemucs"})["result"]["backend"] == "demucs"
+    assert cache.get(cid, "stems", {"model": "merge"})["result"]["backend"] == "merge"
+
+
+def test_hits_increment():
+    cid = "cid_hits"
+    cache.put(cid, "emotion", result={"x": 1})
+    cache.get(cid, "emotion"); cache.get(cid, "emotion")
+    row = next(r for r in cache.stats() if r["kind"] == "emotion")
+    assert row["hits"] >= 2
+
+
+def test_alias_resolves():
+    cache.put_alias("yt:abc123", "cid_aliased")
+    assert cache.resolve_alias("yt:abc123") == "cid_aliased"
+    assert cache.resolve_alias("yt:nope") is None
+
+
+def test_gc_evicts_coldest_to_cap():
+    # three big entries; cap forces eviction of the least-recently-accessed
+    for i in range(3):
+        cache.put(f"cid_big_{i}", "stems", result={"i": i}, bytes_=1000)
+    cache.get("cid_big_2", "stems")          # touch #2 → newest access
+    res = cache.gc("stems", max_bytes=1500)  # room for ~1 entry
+    assert res["evicted"] >= 1
+    assert cache.get("cid_big_2", "stems") is not None  # the touched one survives

armada/api/tests/test_smoke.py

@@ -47,17 +47,28 @@ def test_emotion_rejects_bad_token():
     assert r.status_code == 401
 
 
-def test_emotion_happy_path_mocked(monkeypatch):
-    # Stub the heavy CLAP call — we're testing the route/contract, not the model.
+def test_emotion_happy_path_and_cache(monkeypatch):
+    # Stub the heavy CLAP call — we're testing the route/contract + cache, not the model.
     fake = {"valence": 0.42, "arousal": -0.10, "top": [["warm", 0.5]], "confidence": 0.2}
-    monkeypatch.setattr(ears, "emotion_of", lambda _p: fake)
-    r = client.post("/v1/analyze/emotion",
-                    files={"file": ("c.wav", b"RIFF0123456789", "audio/wav")}, headers=AUTH)
-    assert r.status_code == 200
-    body = r.json()
-    assert body["filename"] == "c.wav"
-    assert body["emotion"]["valence"] == 0.42
-    assert -1 <= body["emotion"]["arousal"] <= 1
+    calls = {"n": 0}
+
+    def _fake(_p):
+        calls["n"] += 1
+        return fake
+    monkeypatch.setattr(ears, "emotion_of", _fake)
+
+    payload = b"UNIQUE-cache-probe-bytes-9f3a"   # undecodable → raw_ content id
+    files = {"file": ("c.wav", payload, "audio/wav")}
+
+    r1 = client.post("/v1/analyze/emotion", files=files, headers=AUTH)
+    assert r1.status_code == 200
+    assert r1.headers["X-Douanier-Cache"] == "miss"
+    assert r1.json()["emotion"]["valence"] == 0.42
+    assert r1.json()["content_id"].startswith(("cid_", "raw_"))
+
+    r2 = client.post("/v1/analyze/emotion", files=files, headers=AUTH)
+    assert r2.headers["X-Douanier-Cache"] == "hit"
+    assert calls["n"] == 1            # engine ran ONCE; 2nd served from cache
 
 
 def test_emotion_too_large(monkeypatch):