feat(audio-api): /sources (#29) + artifact store (#27)

Two links of the heavy chain, built on the #25 job backbone. Both ship the CODE
now; they go live the moment erable ssh is back (yt-dlp install) — no GPU needed.

#27 — artifact store (artifacts.py). Big binaries (fetched sources, later stems/
loops) live content-addressed on local disk under FOURIER_ARTIFACTS, tracked as
ordinary cache rows (kind→result_ref), so the existing LRU cache.gc() already
evicts the coldest under a cap. Freebox was the original SSOT plan — deferred per
PLN; erable-local for now. Serving is zero-copy via nginx X-Accel-Redirect
(FOURIER_X_ACCEL), FileResponse fallback in dev. resolve() refuses path traversal
/ anything outside the root.

#29 — /sources (engines/sources.py + POST /sources). The worker shells out to
yt-dlp → bestaudio → 44.1k WAV (analysis-ready for /separate, /loops, features),
content-addresses it, stores it, and aliases url→content_id so a repeat URL is an
idempotent born-done job (no re-download). Submit returns 202 {job_id}; poll
/jobs/{id}. URL scope is OPEN (any http(s)) per PLN — gated by the platform apikey
+ nechapi monitoring — but we still hard-block the SSRF footguns (non-http(s),
localhost, RFC1918/link-local/reserved IPs; incl. the 169.254.169.254 metadata
classic). The fetch is a thin seam so tests mock the network with a synthetic WAV.

Also: GET /artifacts/{cid}/{name} serving endpoint; healthz reports yt-dlp
presence; _meter now counts by ROUTE TEMPLATE not concrete path (else /jobs/{id}
& /artifacts/{id} would mint unbounded Prometheus series). OpenAPI refreshed
13→17 ops, @nech/api regenerated (submitSource/getArtifact/getJob/cancelJob).
yt-dlp added to requirements-deploy; DEPLOY.md §5b documents the yt-dlp+ffmpeg
install and the nginx internal location. 78→95 tests green.

armada/api/DEPLOY.md

@@ -51,6 +51,10 @@ mkdir -p /srv/fourier/data        # SRE-owned; cap at 2 GB (see §5)
 # image default FOURIER_EMOTION_ENGINE=light is correct for erable — leave unset
 # image default FOURIER_ROOT_PATH=/audio/v1 matches the gateway — leave unset
 FOURIER_MAX_UPLOAD_MB=40
+# artifact store + /sources (#27/#29) — see §5b:
+FOURIER_ARTIFACTS_MAX_MB=1024     # LRU cap for fetched sources / stems / loops
+FOURIER_X_ACCEL=1                 # nginx streams artifacts (needs the §5b location)
+# FOURIER_FETCH_MAX_MB / FOURIER_FETCH_TIMEOUT cap each yt-dlp download (defaults 60/180)
 # Single-thread BLAS on erable's old kernel (see §4 gotcha #2):
 OPENBLAS_NUM_THREADS=1
 OMP_NUM_THREADS=1
@@ -110,6 +114,35 @@ the host dir:
 docker exec fourier python fourier.py cache gc --max-mb 2048
 ```
 
+## 5b. /sources fetch + artifact serving (#29 / #27)
+
+`/sources` fetches a URL's audio in the **worker** (not the API container) and
+stores it as a content-addressed WAV under `FOURIER_ARTIFACTS`
+(default `<cache>/artifacts`, LRU-capped by `FOURIER_ARTIFACTS_MAX_MB`, 1 GB).
+The worker host needs two binaries on PATH:
+
+```bash
+# on the box that runs the worker (erable):
+python -m pip install yt-dlp          # or pin via requirements-deploy.txt
+apt-get install -y ffmpeg             # yt-dlp shells out to it for the WAV transcode
+```
+
+Artifacts are served **zero-copy by nginx**, not the Python process. Set
+`FOURIER_X_ACCEL=1` in the env file and add the internal location to the vhost
+(SRE-owned; the app emits `X-Accel-Redirect: /_artifacts/<cid>/<name>`):
+
+```nginx
+# internal: only reachable via an X-Accel-Redirect from the app, never directly
+location /_artifacts/ {
+    internal;
+    alias /srv/fourier/data/artifacts/;   # = the container's FOURIER_ARTIFACTS
+}
+```
+
+Without `FOURIER_X_ACCEL` the app streams a `FileResponse` itself (fine for dev,
+not for the box). `GET /audio/v1/artifacts/{cid}/{name}` is auth-gated (any tenant
+holding the opaque ref); the bytes are Range-capable through nginx.
+
 ## 6. Observability
 
 `GET /metrics` (internal) / `https://api.nech.pl/audio/v1/metrics` exposes

armada/api/README.md

@@ -63,6 +63,8 @@ Endpoints (internal root paths shown; public = `/audio/v1` + path):
 - `POST /loudness` — audio → BS.1770 integrated LUFS + sample/true peak + crest + gain-to-target (-14/-9) (scope `loudness`)
 - `POST /spectrum` — **FFT-as-a-service**: downsampled spectrogram `bands`×`frames`, 0..1; `?mel=` `?bands=` `?frames=` (`frames=1` = avg spectrum) (scope `spectrum`)
 - `POST /naming` — sample → convention-compliant name (NN_role_character) from MEASURED role+character; `?index=` (scope `naming`)
+- `POST /sources` — `{url}` → fetch best audio as a **job** → 44.1k WAV; `202 {job_id}`, poll `/jobs/{id}` for `{content_id, artifact_ref}`. Idempotent: a re-fetched URL is born-`done` (scope `sources`) (#29)
+- `GET  /artifacts/{cid}/{name}` — stream a stored artifact (fetched source, later stems/loops); nginx `X-Accel-Redirect` in prod, `FileResponse` in dev (any authed tenant) (#27)
 - `POST /separate` — stem separation; **503 `compute_unavailable`** until a GPU runner (scope `separate`)
 
 All three analyze routes are **torch-free** (librosa) and **content-addressed
@@ -131,11 +133,14 @@ hit`), **torch-free light V/A engine + CPU Docker image + platform contract
 image is ~1 GB and the SRE edge is live and waiting.
 
 Also done: OpenAPI client (#32→#44, generated `@nech/api`), hermetic tests + CI
-(#35), **async job backbone + worker (#25)** — the `jobs`/`worker` modules,
-`GET`/`DELETE /jobs/{id}`, atomic claim, crash-recovery, 78 tests green.
-
-Superseded by the platform: our own SQLite bearer auth (#23) and per-token quota
-(#24) — central auth/metering owns these now (the `auth.py` token store + CLI
-remain for local dev only). Queued *under* the public shape without changing it:
-artifact store + X-Accel-Redirect (#27), `/sources`→`/separate`→`/loops` heavy
-chain (#29–#31, each registers a worker handler).
+(#35), **async job backbone + worker (#25)**, **artifact store (#27)** —
+content-addressed local disk, LRU-evicted, served via nginx `X-Accel-Redirect` —
+and **`/sources` (#29)** — yt-dlp URL→44.1k-WAV fetch as a worker job, idempotent
+per URL. **95 tests green.**
+
+Pending the box: `/sources` needs `yt-dlp` installed on the worker host (erable),
+and the artifacts `X-Accel-Redirect` needs the nginx internal location wired (see
+`DEPLOY.md`). Superseded by the platform: our own SQLite bearer auth (#23) and
+per-token quota (#24) — central auth/metering owns these now (the `auth.py` token
+store + CLI remain for local dev only). Still queued *under* the public shape:
+`/separate` (#30, needs a GPU runner) → `/loops`/`/correlate` (#31).

armada/api/app.py

@@ -14,8 +14,10 @@ from pathlib import Path
 
 from fastapi import FastAPI, APIRouter, UploadFile, File, Request, HTTPException, Depends, Response
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import FileResponse
 from pydantic import BaseModel
 
+import artifacts
 import auth
 import cache
 import config
@@ -25,6 +27,7 @@ from engines import ears, feats, signal
 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)
+from engines import sources as sources_eng  # registers the "sources" job handler (#29)
 
 app = FastAPI(
     title="Nech.PL Audio Intelligence API",
@@ -56,7 +59,11 @@ _REQ_COUNTS: dict[tuple[str, int], int] = {}
 @app.middleware("http")
 async def _meter(request: Request, call_next):
     resp = await call_next(request)
-    key = (request.url.path, resp.status_code)
+    # Count by the ROUTE TEMPLATE, not the concrete path — else /jobs/{id} and
+    # /artifacts/{id}/{name} would mint an unbounded Prometheus series per id.
+    route = request.scope.get("route")
+    path = getattr(route, "path_format", None) or request.url.path
+    key = (path, resp.status_code)
     _REQ_COUNTS[key] = _REQ_COUNTS.get(key, 0) + 1
     return resp
 
@@ -135,11 +142,19 @@ def healthz():
             "loudness": {"available": True, "hint": "BS.1770 LUFS + peak/true-peak (pyloudnorm, no torch)"},
             "spectrum": {"available": ok, "hint": "FFT-as-a-service: downsampled spectrogram (no torch)"},
             "naming": {"available": ok, "hint": "convention-compliant sample namer (no torch)"},
+            "sources": {"available": _ytdlp_ok(), "hint": "yt-dlp URL fetch → 44.1k WAV (worker job)"},
             "separate": {"available": False, "hint": "no CPU path on erable — GPU runner pending"},
         },
     }
 
 
+def _ytdlp_ok() -> bool:
+    """Is the yt-dlp binary on PATH? (The /sources worker needs it; the API box
+    may not have it — healthz says so honestly.)"""
+    import shutil
+    return shutil.which(config.YTDLP_BIN) is not None
+
+
 @app.get("/metrics", include_in_schema=False)
 def metrics():
     """Prometheus text exposition (the erable scraper hits this). Counts only —
@@ -214,6 +229,54 @@ def cancel_job(job_id: str, principal: auth.Principal = Depends(require_auth)):
     return jobslib.get(job_id, account=principal.account)
 
 
+# ── /sources (#29): fetch a URL's audio as a job; idempotent on repeat ──────────
+class SourceRequest(BaseModel):
+    """Submit a URL for audio fetch. `url` is any http(s) URL yt-dlp supports."""
+    url: str
+    priority: int = 0
+    webhook_url: str | None = None
+
+
+@v1.post("/sources", operation_id="submitSource", status_code=202, response_model=JobStatus)
+def submit_source(req: SourceRequest, principal: auth.Principal = Depends(require)):
+    """Fetch a URL's best audio → 44.1k WAV, as an async job. Returns `202` with a
+    `job_id`; poll `GET /jobs/{id}` for the source descriptor (content_id +
+    artifact_ref). A URL we've already fetched (and whose artifact still exists)
+    comes back as a born-`done` job immediately — no re-download."""
+    try:
+        url = sources_eng.normalize_url(req.url)
+    except sources_eng.SourceError as e:
+        raise HTTPException(422, f"bad source url: {e}")
+
+    cid = cache.resolve_alias(url)                     # seen this URL before?
+    if cid:
+        hit = cache.get(cid, "source")
+        if hit and hit["result_ref"] and Path(hit["result_ref"]).exists():
+            return jobslib.enqueue(principal.account, "sources", {"url": url},
+                                   result={**hit["result"], "cached": True})
+    return jobslib.enqueue(principal.account, "sources", {"url": url},
+                           priority=req.priority, webhook_url=req.webhook_url)
+
+
+# ── /artifacts (#27): serve a stored binary; nginx X-Accel-Redirect in prod ─────
+@v1.get("/artifacts/{content_id}/{name}", operation_id="getArtifact")
+def get_artifact(content_id: str, name: str,
+                 _p: auth.Principal = Depends(require_auth)):
+    """Stream a stored artifact (a fetched source WAV, later stems/loops). Any
+    authenticated tenant holding the (opaque, content-addressed) ref may fetch it.
+    In prod the app returns an `X-Accel-Redirect` and nginx streams the bytes
+    (Range-capable, zero-copy); in dev it falls back to a direct FileResponse."""
+    path = artifacts.resolve(content_id, name)
+    if path is None:
+        raise HTTPException(404, "artifact not found")
+    if config.X_ACCEL:
+        return Response(status_code=200, headers={
+            "X-Accel-Redirect": artifacts.accel_path(content_id, name),
+            "Content-Type": "application/octet-stream",
+        })
+    return FileResponse(path)
+
+
 # ── shared analyze flow: upload → content-address → cache → compute (#26) ───────
 async def _read_upload(file: UploadFile) -> bytes:
     data = await file.read()

armada/api/artifacts.py

+"""Artifact store (#27) — content-addressed big binaries on local disk.
+
+Fetched sources (#29), separated stems (#30) and loops (#31) are large and
+rebuildable, so they don't belong inline in SQLite. They live on local disk
+under `config.ARTIFACTS_ROOT/<content_id>/<name>` and are tracked as ordinary
+`cache` rows (kind → `result_ref` = path, `bytes` = size), which means the
+existing LRU `cache.gc()` already evicts the coldest ones to keep the footprint
+under a cap. (Freebox was the original SSOT plan; deferred — erable-local for now.)
+
+Serving is **zero-copy via nginx**: the app responds with an `X-Accel-Redirect`
+header pointing at an internal nginx location mapped to ARTIFACTS_ROOT, and nginx
+streams the bytes (Range-capable) without the Python process touching them. In
+dev (no nginx) we fall back to a `FileResponse`. Toggle with `FOURIER_X_ACCEL`.
+"""
+from __future__ import annotations
+
+import shutil
+from pathlib import Path
+
+import cache
+import config
+
+
+def _dir_for(content_id: str) -> Path:
+    d = config.ARTIFACTS_ROOT / content_id
+    d.mkdir(parents=True, exist_ok=True)
+    return d
+
+
+def store(content_id: str, name: str, src_path: str | Path, *, kind: str) -> str:
+    """Move `src_path` into the content-addressed store and return its on-disk ref.
+    The caller registers the ref in the cache table (so eviction tracks it)."""
+    dest = _dir_for(content_id) / Path(name).name        # basename only — no traversal
+    shutil.move(str(src_path), str(dest))                # handles cross-device temp→store
+    return str(dest)
+
+
+def resolve(content_id: str, name: str) -> Path | None:
+    """Map a (content_id, name) request to a real file, refusing path traversal and
+    anything outside ARTIFACTS_ROOT. Returns None if it doesn't exist."""
+    root = config.ARTIFACTS_ROOT.resolve()
+    try:
+        p = (root / content_id / Path(name).name).resolve()
+        p.relative_to(root)                              # raises if escaped the root
+    except (ValueError, OSError):
+        return None
+    return p if p.is_file() else None
+
+
+def accel_path(content_id: str, name: str) -> str:
+    """The internal nginx location the X-Accel-Redirect header points at."""
+    return f"{config.X_ACCEL_LOCATION.rstrip('/')}/{content_id}/{Path(name).name}"
+
+
+def enforce_cap(kind: str, cap_mb: int | None = None) -> dict:
+    """LRU-evict `kind` artifacts until under the cap (delegates to cache.gc, which
+    also unlinks the result_ref files). Returns {evicted, freed}."""
+    cap = (cap_mb if cap_mb is not None else config.ARTIFACTS_MAX_MB)
+    return cache.gc(kind, cap * 1024 * 1024)

armada/api/clients/nech-api/src/audio/apis/AudioApi.ts

@@ -23,6 +23,11 @@ import {
     JobStatusFromJSON,
     JobStatusToJSON,
 } from '../models/JobStatus';
+import {
+    type SourceRequest,
+    SourceRequestFromJSON,
+    SourceRequestToJSON,
+} from '../models/SourceRequest';
 
 export interface AnalyzeAllRequest {
     file: Blob;
@@ -46,6 +51,11 @@ export interface FeaturesRequest {
     rhythm?: boolean;
 }
 
+export interface GetArtifactRequest {
+    contentId: string;
+    name: string;
+}
+
 export interface GetJobRequest {
     jobId: string;
 }
@@ -75,6 +85,10 @@ export interface SpectrumRequest {
     mel?: boolean;
 }
 
+export interface SubmitSourceRequest {
+    sourceRequest: SourceRequest;
+}
+
 export interface WaveformRequest {
     file: Blob;
     bins?: number;
@@ -425,6 +439,65 @@ export class AudioApi extends runtime.BaseAPI {
     }
 
     /**
+     * Creates request options for getArtifact without sending the request
+     */
+    async getArtifactRequestOpts(requestParameters: GetArtifactRequest): Promise<runtime.RequestOpts> {
+        if (requestParameters['contentId'] == null) {
+            throw new runtime.RequiredError(
+                'contentId',
+                'Required parameter "contentId" was null or undefined when calling getArtifact().'
+            );
+        }
+
+        if (requestParameters['name'] == null) {
+            throw new runtime.RequiredError(
+                'name',
+                'Required parameter "name" was null or undefined when calling getArtifact().'
+            );
+        }
+
+        const queryParameters: any = {};
+
+        const headerParameters: runtime.HTTPHeaders = {};
+
+
+        let urlPath = `/artifacts/{content_id}/{name}`;
+        urlPath = urlPath.replace('{content_id}', encodeURIComponent(String(requestParameters['contentId'])));
+        urlPath = urlPath.replace('{name}', encodeURIComponent(String(requestParameters['name'])));
+
+        return {
+            path: urlPath,
+            method: 'GET',
+            headers: headerParameters,
+            query: queryParameters,
+        };
+    }
+
+    /**
+     * Stream a stored artifact (a fetched source WAV, later stems/loops). Any authenticated tenant holding the (opaque, content-addressed) ref may fetch it. In prod the app returns an `X-Accel-Redirect` and nginx streams the bytes (Range-capable, zero-copy); in dev it falls back to a direct FileResponse.
+     * Get Artifact
+     */
+    async getArtifactRaw(requestParameters: GetArtifactRequest, initOverrides?: RequestInit | runtime.InitOverrideFunction): Promise<runtime.ApiResponse<any>> {
+        const requestOptions = await this.getArtifactRequestOpts(requestParameters);
+        const response = await this.request(requestOptions, initOverrides);
+
+        if (this.isJsonMime(response.headers.get('content-type'))) {
+            return new runtime.JSONApiResponse<any>(response);
+        } else {
+            return new runtime.TextApiResponse(response) as any;
+        }
+    }
+
+    /**
+     * Stream a stored artifact (a fetched source WAV, later stems/loops). Any authenticated tenant holding the (opaque, content-addressed) ref may fetch it. In prod the app returns an `X-Accel-Redirect` and nginx streams the bytes (Range-capable, zero-copy); in dev it falls back to a direct FileResponse.
+     * Get Artifact
+     */
+    async getArtifact(requestParameters: GetArtifactRequest, initOverrides?: RequestInit | runtime.InitOverrideFunction): Promise<any> {
+        const response = await this.getArtifactRaw(requestParameters, initOverrides);
+        return await response.value();
+    }
+
+    /**
      * Creates request options for getJob without sending the request
      */
     async getJobRequestOpts(requestParameters: GetJobRequest): Promise<runtime.RequestOpts> {
@@ -931,6 +1004,55 @@ export class AudioApi extends runtime.BaseAPI {
     }
 
     /**
+     * Creates request options for submitSource without sending the request
+     */
+    async submitSourceRequestOpts(requestParameters: SubmitSourceRequest): Promise<runtime.RequestOpts> {
+        if (requestParameters['sourceRequest'] == null) {
+            throw new runtime.RequiredError(
+                'sourceRequest',
+                'Required parameter "sourceRequest" was null or undefined when calling submitSource().'
+            );
+        }
+
+        const queryParameters: any = {};
+
+        const headerParameters: runtime.HTTPHeaders = {};
+
+        headerParameters['Content-Type'] = 'application/json';
+
+
+        let urlPath = `/sources`;
+
+        return {
+            path: urlPath,
+            method: 'POST',
+            headers: headerParameters,
+            query: queryParameters,
+            body: SourceRequestToJSON(requestParameters['sourceRequest']),
+        };
+    }
+
+    /**
+     * Fetch a URL\'s best audio → 44.1k WAV, as an async job. Returns `202` with a `job_id`; poll `GET /jobs/{id}` for the source descriptor (content_id + artifact_ref). A URL we\'ve already fetched (and whose artifact still exists) comes back as a born-`done` job immediately — no re-download.
+     * Submit Source
+     */
+    async submitSourceRaw(requestParameters: SubmitSourceRequest, initOverrides?: RequestInit | runtime.InitOverrideFunction): Promise<runtime.ApiResponse<JobStatus>> {
+        const requestOptions = await this.submitSourceRequestOpts(requestParameters);
+        const response = await this.request(requestOptions, initOverrides);
+
+        return new runtime.JSONApiResponse(response, (jsonValue) => JobStatusFromJSON(jsonValue));
+    }
+
+    /**
+     * Fetch a URL\'s best audio → 44.1k WAV, as an async job. Returns `202` with a `job_id`; poll `GET /jobs/{id}` for the source descriptor (content_id + artifact_ref). A URL we\'ve already fetched (and whose artifact still exists) comes back as a born-`done` job immediately — no re-download.
+     * Submit Source
+     */
+    async submitSource(requestParameters: SubmitSourceRequest, initOverrides?: RequestInit | runtime.InitOverrideFunction): Promise<JobStatus> {
+        const response = await this.submitSourceRaw(requestParameters, initOverrides);
+        return await response.value();
+    }
+
+    /**
      * Creates request options for waveform without sending the request
      */
     async waveformRequestOpts(requestParameters: WaveformRequest): Promise<runtime.RequestOpts> {

armada/api/clients/nech-api/src/audio/models/SourceRequest.ts

+/* tslint:disable */
+/* eslint-disable */
+/**
+ * Nech.PL Audio Intelligence API
+ * Stem-aware audio analysis — emotion, features, sample role, loop grading, onsets/tempo, waveform & spectrogram, loudness, naming. The `audio` API of the Nech.PL APIs platform, served at /audio/v1/...; the gateway does central auth + metering (Bearer token).
+ *
+ * The version of the OpenAPI document: 0.1.0
+ * 
+ *
+ * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
+ * https://openapi-generator.tech
+ * Do not edit the class manually.
+ */
+
+import { mapValues } from '../runtime';
+/**
+ * Submit a URL for audio fetch. `url` is any http(s) URL yt-dlp supports.
+ * @export
+ * @interface SourceRequest
+ */
+export interface SourceRequest {
+    /**
+     * 
+     * @type {string}
+     * @memberof SourceRequest
+     */
+    url: string;
+    /**
+     * 
+     * @type {number}
+     * @memberof SourceRequest
+     */
+    priority?: number;
+    /**
+     * 
+     * @type {string}
+     * @memberof SourceRequest
+     */
+    webhookUrl?: string | null;
+}
+
+/**
+ * Check if a given object implements the SourceRequest interface.
+ */
+export function instanceOfSourceRequest(value: object): value is SourceRequest {
+    if (!('url' in value) || value['url'] === undefined) return false;
+    return true;
+}
+
+export function SourceRequestFromJSON(json: any): SourceRequest {
+    return SourceRequestFromJSONTyped(json, false);
+}
+
+export function SourceRequestFromJSONTyped(json: any, ignoreDiscriminator: boolean): SourceRequest {
+    if (json == null) {
+        return json;
+    }
+    return {
+        
+        'url': json['url'],
+        'priority': json['priority'] == null ? undefined : json['priority'],
+        'webhookUrl': json['webhook_url'] == null ? undefined : json['webhook_url'],
+    };
+}
+
+export function SourceRequestToJSON(json: any): SourceRequest {
+    return SourceRequestToJSONTyped(json, false);
+}
+
+export function SourceRequestToJSONTyped(value?: SourceRequest | null, ignoreDiscriminator: boolean = false): any {
+    if (value == null) {
+        return value;
+    }
+
+    return {
+        
+        'url': value['url'],
+        'priority': value['priority'],
+        'webhook_url': value['webhookUrl'],
+    };
+}
+

armada/api/clients/nech-api/src/audio/models/index.ts

@@ -3,4 +3,5 @@
 export * from './HTTPValidationError';
 export * from './JobStatus';
 export * from './LocationInner';
+export * from './SourceRequest';
 export * from './ValidationError';

armada/api/clients/openapi.json

@@ -129,6 +129,98 @@
         }
       }
     },
+    "/sources": {
+      "post": {
+        "tags": [
+          "audio"
+        ],
+        "summary": "Submit Source",
+        "description": "Fetch a URL's best audio → 44.1k WAV, as an async job. Returns `202` with a\n`job_id`; poll `GET /jobs/{id}` for the source descriptor (content_id +\nartifact_ref). A URL we've already fetched (and whose artifact still exists)\ncomes back as a born-`done` job immediately — no re-download.",
+        "operationId": "submitSource",
+        "requestBody": {
+          "content": {
+            "application/json": {
+              "schema": {
+                "$ref": "#/components/schemas/SourceRequest"
+              }
+            }
+          },
+          "required": true
+        },
+        "responses": {
+          "202": {
+            "description": "Successful Response",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/JobStatus"
+                }
+              }
+            }
+          },
+          "422": {
+            "description": "Validation Error",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/HTTPValidationError"
+                }
+              }
+            }
+          }
+        }
+      }
+    },
+    "/artifacts/{content_id}/{name}": {
+      "get": {
+        "tags": [
+          "audio"
+        ],
+        "summary": "Get Artifact",
+        "description": "Stream a stored artifact (a fetched source WAV, later stems/loops). Any\nauthenticated tenant holding the (opaque, content-addressed) ref may fetch it.\nIn prod the app returns an `X-Accel-Redirect` and nginx streams the bytes\n(Range-capable, zero-copy); in dev it falls back to a direct FileResponse.",
+        "operationId": "getArtifact",
+        "parameters": [
+          {
+            "name": "content_id",
+            "in": "path",
+            "required": true,
+            "schema": {
+              "type": "string",
+              "title": "Content Id"
+            }
+          },
+          {
+            "name": "name",
+            "in": "path",
+            "required": true,
+            "schema": {
+              "type": "string",
+              "title": "Name"
+            }
+          }
+        ],
+        "responses": {
+          "200": {
+            "description": "Successful Response",
+            "content": {
+              "application/json": {
+                "schema": {}
+              }
+            }
+          },
+          "422": {
+            "description": "Validation Error",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/HTTPValidationError"
+                }
+              }
+            }
+          }
+        }
+      }
+    },
     "/analyze/emotion": {
       "post": {
         "tags": [
@@ -887,6 +979,36 @@
         "title": "JobStatus",
         "description": "The status of an async job (separate/sources/loops). `result` is populated\nonce `status == \"done\"`; `error` once `status == \"error\"`."
       },
+      "SourceRequest": {
+        "properties": {
+          "url": {
+            "type": "string",
+            "title": "Url"
+          },
+          "priority": {
+            "type": "integer",
+            "title": "Priority",
+            "default": 0
+          },
+          "webhook_url": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Webhook Url"
+          }
+        },
+        "type": "object",
+        "required": [
+          "url"
+        ],
+        "title": "SourceRequest",
+        "description": "Submit a URL for audio fetch. `url` is any http(s) URL yt-dlp supports."
+      },
       "ValidationError": {
         "properties": {
           "loc": {

armada/api/config.py

@@ -44,6 +44,26 @@ DEV_TOKEN = os.environ.get("FOURIER_DEV_TOKEN", "")
 # ── limits ─────────────────────────────────────────────────────────────────────
 MAX_UPLOAD_MB = int(os.environ.get("FOURIER_MAX_UPLOAD_MB", "40"))
 
+# ── artifact store (#27) — local disk on erable; freebox deferred ───────────────
+# Big binaries (fetched sources, stems, loops) live content-addressed under here,
+# tracked as cache rows (kind→result_ref) and LRU-evicted to stay under the cap.
+ARTIFACTS_ROOT = Path(os.environ.get("FOURIER_ARTIFACTS", CACHE_ROOT / "artifacts"))
+ARTIFACTS_MAX_MB = int(os.environ.get("FOURIER_ARTIFACTS_MAX_MB", "1024"))   # 1 GB cap
+# Serving: in prod nginx sends the bytes via X-Accel-Redirect (the app never
+# streams big files itself). The header points at this INTERNAL nginx location,
+# which maps to ARTIFACTS_ROOT. Off by default → dev serves a FileResponse.
+X_ACCEL = os.environ.get("FOURIER_X_ACCEL", "").lower() in ("1", "true", "yes")
+X_ACCEL_LOCATION = os.environ.get("FOURIER_X_ACCEL_LOCATION", "/_artifacts")
+
+# ── /sources fetch (#29) — yt-dlp as a worker job ───────────────────────────────
+# The worker is the ONLY process that shells out to yt-dlp. URL scope is OPEN
+# (any http(s)), gated by the platform apikey + nechapi monitoring; we still hard-
+# block SSRF footguns (file://, localhost, RFC1918/link-local) in sources.py.
+YTDLP_BIN = os.environ.get("FOURIER_YTDLP", "yt-dlp")
+FETCH_TIMEOUT_SEC = int(os.environ.get("FOURIER_FETCH_TIMEOUT", "180"))
+FETCH_MAX_MB = int(os.environ.get("FOURIER_FETCH_MAX_MB", "60"))   # per-download cap
+FETCH_SR = int(os.environ.get("FOURIER_FETCH_SR", "44100"))        # transcode target
+
 # ── engine selection ─────────────────────────────────────────────────────────
 # "clap"  = CLAP/torch (rich, ~4 GB, needs the GPU/dev box) — default for local dev.
 # "light" = librosa-heuristic V/A (no torch, fits the erable container) — set in prod.

armada/api/engines/sources.py

+"""/sources (#29) — fetch a URL's audio as a worker job.
+
+The worker is the ONLY process that shells out to **yt-dlp**: it downloads the
+best audio stream of a URL, transcodes to 44.1 kHz WAV (analysis-ready for
+/separate, /loops, the feature stack), content-addresses it, stores it as a
+`source` artifact (#27), and aliases the URL → content_id so a repeat fetch is
+free (the submit path turns that into an idempotent born-done job).
+
+URL scope is OPEN — any http(s) URL yt-dlp supports — gated upstream by the
+platform apikey + nechapi monitoring (trusted-tenant model). We still hard-block
+the SSRF footguns here: non-http(s) schemes, localhost, and RFC1918/link-local/
+reserved IP literals. (DNS-rebinding to an internal name is out of scope for the
+MVP; the apikey gate carries that risk.)
+
+Heavy deps (yt-dlp, ffmpeg) load only in the worker. `_fetch_audio_wav` is a thin
+seam so tests monkeypatch the network call with a synthetic WAV.
+"""
+from __future__ import annotations
+
+import ipaddress
+import subprocess
+from pathlib import Path
+from urllib.parse import urlparse, urlunparse
+
+import artifacts
+import cache
+import config
+import jobs
+
+
+class SourceError(ValueError):
+    """A caller-fixable problem (bad/blocked URL) — surfaced as 422 at submit."""
+
+
+# ── URL validation / normalization ──────────────────────────────────────────────
+def normalize_url(raw: str) -> str:
+    """Validate + canonicalize a fetch URL. Raises SourceError on a rejected URL.
+    The canonical form is the alias key, so trivially-different URLs to the same
+    resource still dedupe (lowercased host, no fragment)."""
+    raw = (raw or "").strip()
+    if not raw:
+        raise SourceError("empty url")
+    u = urlparse(raw)
+    if u.scheme not in ("http", "https"):
+        raise SourceError(f"unsupported scheme '{u.scheme}' (only http/https)")
+    host = (u.hostname or "").lower()
+    if not host:
+        raise SourceError("url has no host")
+    if host in ("localhost", "localhost.localdomain") or host.endswith(".localhost"):
+        raise SourceError("refusing localhost")
+    try:
+        ip = ipaddress.ip_address(host)                 # host is a literal IP?
+    except ValueError:
+        ip = None
+    if ip is not None and (ip.is_private or ip.is_loopback or ip.is_link_local
+                           or ip.is_reserved or ip.is_multicast or ip.is_unspecified):
+        raise SourceError(f"refusing non-public address {host}")
+    # canonical: drop fragment, lowercase host, keep path+query (e.g. ?v=…)
+    return urlunparse((u.scheme, host + (f":{u.port}" if u.port else ""),
+                       u.path, u.params, u.query, ""))
+
+
+# ── the fetch seam (monkeypatched in tests) ──────────────────────────────────────
+def _fetch_audio_wav(url: str, outdir: str) -> Path:
+    """Download best audio → 44.1k WAV via yt-dlp+ffmpeg. Returns the WAV path."""
+    out_tmpl = str(Path(outdir) / "src.%(ext)s")
+    cmd = [
+        config.YTDLP_BIN, "--quiet", "--no-playlist", "--no-progress",
+        "-f", "bestaudio/best",
+        "-x", "--audio-format", "wav",
+        "--postprocessor-args", f"ffmpeg:-ar {config.FETCH_SR}",
+        "--max-filesize", f"{config.FETCH_MAX_MB}M",
+        "--socket-timeout", "30",
+        "-o", out_tmpl, url,
+    ]
+    try:
+        subprocess.run(cmd, check=True, timeout=config.FETCH_TIMEOUT_SEC,
+                       capture_output=True, text=True)
+    except FileNotFoundError as e:
+        raise RuntimeError(f"yt-dlp not installed ({config.YTDLP_BIN})") from e
+    except subprocess.TimeoutExpired as e:
+        raise RuntimeError(f"fetch timed out after {config.FETCH_TIMEOUT_SEC}s") from e
+    except subprocess.CalledProcessError as e:
+        tail = (e.stderr or "").strip().splitlines()[-1:] or ["(no stderr)"]
+        raise RuntimeError(f"yt-dlp failed: {tail[0]}") from e
+    wavs = sorted(Path(outdir).glob("*.wav"))
+    if not wavs:
+        raise RuntimeError("yt-dlp produced no audio (no extractable stream?)")
+    return wavs[0]
+
+
+def _probe(wav: Path) -> dict:
+    """duration / samplerate / channels / bytes of the produced WAV."""
+    import soundfile as sf
+    info = sf.info(str(wav))
+    return {
+        "duration_sec": round(info.frames / info.samplerate, 3) if info.samplerate else 0.0,
+        "sample_rate": info.samplerate,
+        "channels": info.channels,
+        "bytes": wav.stat().st_size,
+    }
+
+
+# ── the worker handler ────────────────────────────────────────────────────────
+@jobs.handler("sources")
+def run_sources(job: dict, progress) -> dict:
+    """Fetch → transcode → content-address → store → alias. Returns the source
+    descriptor (also cached under the content_id so a repeat URL is born-done)."""
+    import tempfile
+
+    url = job["params"]["url"]
+    progress(0.05)
+    with tempfile.TemporaryDirectory(prefix="fourier-src-") as td:
+        wav = _fetch_audio_wav(url, td)
+        progress(0.7)
+        cid = cache.content_id(wav)
+        meta = _probe(wav)
+        ref = artifacts.store(cid, "source.wav", wav, kind="source")
+
+    result = {
+        "content_id": cid,
+        "artifact_ref": f"{cid}/source.wav",      # the GET /artifacts/{cid}/{name} key
+        "url": url,
+        **{k: v for k, v in meta.items() if k != "bytes"},
+    }
+    cache.put(cid, "source", result=result, result_ref=ref, bytes_=meta["bytes"])
+    cache.put_alias(url, cid)
+    artifacts.enforce_cap("source")               # keep the store under its cap
+    progress(1.0)
+    return result

armada/api/requirements-deploy.txt

@@ -18,3 +18,7 @@ numpy>=1.26,<2.3
 librosa>=0.11               # spectral/chroma/tempo features (pulls scipy, numba, sklearn)
 soundfile>=0.12             # content_id PCM decode + librosa wav/flac backend
 pyloudnorm>=0.1             # BS.1770 integrated loudness (LUFS) for /loudness — pure-python, tiny
+
+# /sources (#29) URL fetch — worker-only. Needs the `ffmpeg` BINARY on the host
+# too (apt: ffmpeg) for the WAV transcode; yt-dlp shells out to it.
+yt-dlp>=2024.0

armada/api/tests/conftest.py

@@ -11,6 +11,9 @@ from pathlib import Path
 _DB = Path(tempfile.gettempdir()) / "fourier_test.db"
 os.environ["FOURIER_DB"] = str(_DB)
 os.environ.setdefault("FOURIER_DEV_TOKEN", "test-secret")
+# keep the artifact store (#27) out of the repo's _cache during tests
+_ART = Path(tempfile.gettempdir()) / "fourier_test_artifacts"
+os.environ["FOURIER_ARTIFACTS"] = str(_ART)
 
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 

armada/api/tests/test_sources.py

+"""/sources (#29) + artifact store (#27): URL validation, the fetch handler
+(yt-dlp seam mocked), idempotent born-done fast path, and artifact serving."""
+import wave
+from pathlib import Path
+
+import pytest
+from fastapi.testclient import TestClient
+
+import app
+import cache
+import config
+import db
+import jobs
+import worker
+from engines import sources as sources_eng
+
+client = TestClient(app.app)
+GW = {"X-Tenant": "hexa", "X-Scopes": "api:audio:*"}
+
+
+@pytest.fixture(autouse=True)
+def _clean():
+    db.init()
+    with db.cursor() as c:
+        c.execute("DELETE FROM jobs")
+        c.execute("DELETE FROM cache")
+        c.execute("DELETE FROM aliases")
+    yield
+
+
+def _write_wav(path: Path, seconds=0.2, sr=44100):
+    """A tiny silent stereo WAV so soundfile.info / content_id work without network."""
+    n = int(seconds * sr)
+    with wave.open(str(path), "wb") as w:
+        w.setnchannels(2)
+        w.setsampwidth(2)
+        w.setframerate(sr)
+        w.writeframes(b"\x00\x00\x00\x00" * n)
+
+
+@pytest.fixture
+def fake_fetch(monkeypatch):
+    """Replace the yt-dlp network call with a synthetic WAV writer."""
+    def _fake(url, outdir):
+        p = Path(outdir) / "src.wav"
+        _write_wav(p)
+        return p
+    monkeypatch.setattr(sources_eng, "_fetch_audio_wav", _fake)
+    return _fake
+
+
+# ── URL validation ───────────────────────────────────────────────────────────
+@pytest.mark.parametrize("url", [
+    "https://www.youtube.com/watch?v=abc123",
+    "http://soundcloud.com/artist/track",
+])
+def test_normalize_accepts_public_http(url):
+    assert sources_eng.normalize_url(url).startswith(("http://", "https://"))
+
+
+def test_normalize_lowercases_host_and_drops_fragment():
+    out = sources_eng.normalize_url("https://YouTube.com/watch?v=x#t=30")
+    assert out == "https://youtube.com/watch?v=x"
+
+
+@pytest.mark.parametrize("bad", [
+    "file:///etc/passwd",
+    "ftp://example.com/x",
+    "http://localhost/x",
+    "http://127.0.0.1/x",
+    "http://10.0.0.5/x",
+    "http://169.254.169.254/latest/meta-data",   # the classic cloud-metadata SSRF
+    "",
+])
+def test_normalize_rejects_ssrf_and_nonhttp(bad):
+    with pytest.raises(sources_eng.SourceError):
+        sources_eng.normalize_url(bad)
+
+
+# ── the worker handler (fetch seam mocked) ──────────────────────────────────────
+def test_handler_fetches_stores_and_aliases(fake_fetch):
+    jid = jobs.enqueue("hexa", "sources",
+                       {"url": "https://youtube.com/watch?v=abc"})["job_id"]
+    worker.run_one(jobs.claim())
+
+    done = jobs.get(jid)
+    assert done["status"] == "done"
+    res = done["result"]
+    assert res["content_id"].startswith(("cid_", "raw_"))
+    assert res["artifact_ref"] == f"{res['content_id']}/source.wav"
+    assert res["sample_rate"] == 44100 and res["channels"] == 2
+
+    # alias registered → URL resolves to the content id; artifact on disk
+    assert cache.resolve_alias("https://youtube.com/watch?v=abc") == res["content_id"]
+    assert (config.ARTIFACTS_ROOT / res["content_id"] / "source.wav").is_file()
+
+
+# ── submit endpoint ──────────────────────────────────────────────────────────
+def test_submit_returns_202_pending_job():
+    r = client.post("/sources", headers=GW,
+                    json={"url": "https://youtube.com/watch?v=zzz"})
+    assert r.status_code == 202
+    body = r.json()
+    assert body["type"] == "sources" and body["status"] == "pending"
+
+
+def test_submit_rejects_bad_url_with_422():
+    r = client.post("/sources", headers=GW, json={"url": "file:///etc/passwd"})
+    assert r.status_code == 422
+
+
+def test_submit_is_idempotent_born_done_after_first_fetch(fake_fetch):
+    url = "https://youtube.com/watch?v=dedupe"
+    # first submit → run it through the worker
+    first = client.post("/sources", headers=GW, json={"url": url}).json()
+    worker.run_one(jobs.claim())
+    assert jobs.get(first["job_id"])["status"] == "done"
+
+    # second submit of the same URL → born done immediately, no new pending job
+    second = client.post("/sources", headers=GW, json={"url": url}).json()
+    assert second["status"] == "done"
+    assert second["result"]["cached"] is True
+    assert jobs.claim() is None            # nothing queued for the worker
+
+
+# ── artifact serving (#27) ──────────────────────────────────────────────────────
+def test_artifact_download_and_guards(fake_fetch):
+    jobs.enqueue("hexa", "sources", {"url": "https://youtube.com/watch?v=art"})
+    worker.run_one(jobs.claim())
+    cid = cache.resolve_alias("https://youtube.com/watch?v=art")
+
+    r = client.get(f"/artifacts/{cid}/source.wav", headers=GW)
+    assert r.status_code == 200 and r.content[:4] == b"RIFF"   # a real WAV
+
+    assert client.get(f"/artifacts/{cid}/missing.wav", headers=GW).status_code == 404
+    # path traversal is refused (resolve() rejects → 404, never escapes the root)
+    assert client.get("/artifacts/..%2f..%2fetc/passwd", headers=GW).status_code == 404
+
+
+def test_artifact_uses_x_accel_when_enabled(fake_fetch, monkeypatch):
+    jobs.enqueue("hexa", "sources", {"url": "https://youtube.com/watch?v=accel"})
+    worker.run_one(jobs.claim())
+    cid = cache.resolve_alias("https://youtube.com/watch?v=accel")
+
+    monkeypatch.setattr(config, "X_ACCEL", True)
+    r = client.get(f"/artifacts/{cid}/source.wav", headers=GW)
+    assert r.status_code == 200
+    assert r.headers["X-Accel-Redirect"] == f"{config.X_ACCEL_LOCATION}/{cid}/source.wav"
+
+
+# ── healthz reflects yt-dlp presence ────────────────────────────────────────────
+def test_healthz_reports_sources_engine():
+    eng = client.get("/healthz").json()["engines"]
+    assert "sources" in eng and isinstance(eng["sources"]["available"], bool)