feat(audio-api): async job backbone + worker daemon (#25)

The async spine for the heavy chain (sources/separate/loops): the API enqueues
and returns a job_id; a worker runs it out-of-band so the API never blocks.
Built to the four decisions taken with PLN:

- HYBRID sync/async: the 11 cheap analyses stay synchronous (cache-backed);
  only heavy work becomes a job. The line is wall-clock cost, not endpoint kind.
- PER-ENGINE submit + SHARED poll: engines return 202 {job_id} (wired in
  #29/#30/#31); GET /jobs/{id} polls, DELETE /jobs/{id} cancels. Typed JobStatus
  response_model so the generated client gets audio.getJob()/cancelJob().
- SINGLE serialized worker: one job at a time (erable is 2 GB / no GPU; two
  demucs runs would OOM). The atomic claim is still race-safe for N workers.
- Identity = gateway tenant; a job is visible only to its owner (cross-tenant
  poll/cancel → 404, not 403, so existence doesn't leak).

Pieces:
- db.py: jobs table (status/priority/progress/result/attempts/webhook) + a
  claim-ordered index.
- jobs.py: enqueue (incl. born-done idempotent fast path when the cache already
  has the result), atomic claim (candidate → guarded UPDATE WHERE status=
  'pending'; rowcount-0 retry; WAL serializes writers so no double-claim),
  progress/finish/fail-with-requeue-under-cap, crash recovery (running→pending
  on boot), cooperative cancel + is_cancelled checkpoints.
- worker.py: serialized loop — recover → claim → dispatch by type → finish/fail
  → optional webhook; SIGTERM-graceful; imports engine handler modules (none yet,
  idles politely); engines register via @jobs.handler.
- deploy/douanier-worker.service: systemd --user unit (linger) — the durable run
  path (harness/nohup jobs die on teardown). Prod container-vs-host wiring is the
  paved-road call (SRE/#34).

VALIDATION: tests/test_jobs.py — submit→claim→done; born-done fast path;
finish/requeue-to-cap; crash recovery; NO double-claim across 6 threads × 25
jobs (each claimed exactly once); worker dispatch to done; missing-handler error;
exception→requeue; cancel mid-flight; HTTP poll + ownership 404 + idempotent
cancel. 11 new tests, full suite 67→78 green. OpenAPI snapshot refreshed (15
ops) + @nech/api regenerated (getJob/cancelJob); noImplicitAny relaxed for the
100%-generated client (typescript-fetch's camel/snake guard trips TS7053).

armada/api/README.md

@@ -98,6 +98,23 @@ docker run -d -p 127.0.0.1:9780:9780 -v /srv/douanier/data:/data douanier:latest
 curl -s https://api.nech.pl/audio/v1/healthz # green once SRE wires the keep-alive
 ```
 
+## Async jobs (#25)
+
+Cheap analyses (emotion, features, …) stay **synchronous** (cache-backed). Heavy
+work (sources / separate / loops) runs as a **job**: the engine endpoint returns
+`202 {job_id}`, the caller polls `GET /jobs/{id}` and (optionally) cancels with
+`DELETE /jobs/{id}`. One serialized worker (erable is 2 GB / no GPU) claims jobs
+FIFO+priority, atomically (race-safe for N workers), with crash-recovery and an
+attempt cap. Identity is the gateway tenant; a job is only visible to its owner.
+
+```bash
+# run the worker (dev); prod = the systemd --user unit in deploy/
+cd armada/api && ~/.virtualenvs/douanier/bin/python -m worker
+```
+
+Engines register a handler with `@jobs.handler("separate")`; the worker is the
+only process that loads the heavy deps. See `jobs.py` / `worker.py`.
+
 ## Test
 
 ```bash
@@ -112,8 +129,12 @@ hit`), **torch-free light V/A engine + CPU Docker image + platform contract
 (`/audio/v1`, central-auth headers, `/metrics`, `/separate` 503) (#37)**. The
 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:
-async jobs + worker (#25), signed artifact URLs (#27), `/features`+`/samples`
-(#28), `/sources`→`/separate`→`/loops` heavy chain (#29–#31), OpenAPI client (#32).
+artifact store + X-Accel-Redirect (#27), `/sources`→`/separate`→`/loops` heavy
+chain (#29–#31, each registers a worker handler).

armada/api/app.py

@@ -14,10 +14,12 @@ from pathlib import Path
 
 from fastapi import FastAPI, APIRouter, UploadFile, File, Request, HTTPException, Depends, Response
 from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
 
 import auth
 import cache
 import config
+import jobs as jobslib
 import scopes as scopelib
 from engines import ears, feats, signal
 from engines import loudness as loudness_eng
@@ -176,6 +178,42 @@ def whoami(principal: auth.Principal = Depends(require_auth)):
             "expires_at": principal.expires_at}
 
 
+# ── async jobs (#25): heavy work runs out-of-band; poll/cancel here ─────────────
+class JobStatus(BaseModel):
+    """The status of an async job (separate/sources/loops). `result` is populated
+    once `status == "done"`; `error` once `status == "error"`."""
+    job_id: str
+    type: str
+    status: str                       # pending | running | done | error | cancelled
+    progress: float                   # 0.0 … 1.0
+    result: dict | None = None
+    result_ref: str | None = None
+    error: str | None = None
+    created: float
+    started: float | None = None
+    finished: float | None = None
+
+
+@v1.get("/jobs/{job_id}", operation_id="getJob", response_model=JobStatus)
+def get_job(job_id: str, principal: auth.Principal = Depends(require_auth)):
+    """Poll an async job. 404 if it doesn't exist OR belongs to another tenant
+    (we don't leak job existence across accounts)."""
+    job = jobslib.get(job_id, account=principal.account)
+    if job is None:
+        raise HTTPException(404, "job not found")
+    return job
+
+
+@v1.delete("/jobs/{job_id}", operation_id="cancelJob", response_model=JobStatus)
+def cancel_job(job_id: str, principal: auth.Principal = Depends(require_auth)):
+    """Cancel a pending/running job (cooperative — a running handler stops at its
+    next progress checkpoint). Idempotent: cancelling a finished job is a no-op."""
+    status = jobslib.cancel(job_id, account=principal.account)
+    if status is None:
+        raise HTTPException(404, "job not found")
+    return jobslib.get(job_id, account=principal.account)
+
+
 # ── shared analyze flow: upload → content-address → cache → compute (#26) ───────
 async def _read_upload(file: UploadFile) -> bytes:
     data = await file.read()

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

@@ -18,6 +18,11 @@ import {
     HTTPValidationErrorFromJSON,
     HTTPValidationErrorToJSON,
 } from '../models/HTTPValidationError';
+import {
+    type JobStatus,
+    JobStatusFromJSON,
+    JobStatusToJSON,
+} from '../models/JobStatus';
 
 export interface AnalyzeAllRequest {
     file: Blob;
@@ -32,11 +37,19 @@ export interface AnalyzeSamplesRequest {
     name?: string;
 }
 
+export interface CancelJobRequest {
+    jobId: string;
+}
+
 export interface FeaturesRequest {
     file: Blob;
     rhythm?: boolean;
 }
 
+export interface GetJobRequest {
+    jobId: string;
+}
+
 export interface GradeRequest {
     file: Blob;
     role?: string;
@@ -290,6 +303,53 @@ export class AudioApi extends runtime.BaseAPI {
     }
 
     /**
+     * Creates request options for cancelJob without sending the request
+     */
+    async cancelJobRequestOpts(requestParameters: CancelJobRequest): Promise<runtime.RequestOpts> {
+        if (requestParameters['jobId'] == null) {
+            throw new runtime.RequiredError(
+                'jobId',
+                'Required parameter "jobId" was null or undefined when calling cancelJob().'
+            );
+        }
+
+        const queryParameters: any = {};
+
+        const headerParameters: runtime.HTTPHeaders = {};
+
+
+        let urlPath = `/jobs/{job_id}`;
+        urlPath = urlPath.replace('{job_id}', encodeURIComponent(String(requestParameters['jobId'])));
+
+        return {
+            path: urlPath,
+            method: 'DELETE',
+            headers: headerParameters,
+            query: queryParameters,
+        };
+    }
+
+    /**
+     * Cancel a pending/running job (cooperative — a running handler stops at its next progress checkpoint). Idempotent: cancelling a finished job is a no-op.
+     * Cancel Job
+     */
+    async cancelJobRaw(requestParameters: CancelJobRequest, initOverrides?: RequestInit | runtime.InitOverrideFunction): Promise<runtime.ApiResponse<JobStatus>> {
+        const requestOptions = await this.cancelJobRequestOpts(requestParameters);
+        const response = await this.request(requestOptions, initOverrides);
+
+        return new runtime.JSONApiResponse(response, (jsonValue) => JobStatusFromJSON(jsonValue));
+    }
+
+    /**
+     * Cancel a pending/running job (cooperative — a running handler stops at its next progress checkpoint). Idempotent: cancelling a finished job is a no-op.
+     * Cancel Job
+     */
+    async cancelJob(requestParameters: CancelJobRequest, initOverrides?: RequestInit | runtime.InitOverrideFunction): Promise<JobStatus> {
+        const response = await this.cancelJobRaw(requestParameters, initOverrides);
+        return await response.value();
+    }
+
+    /**
      * Creates request options for features without sending the request
      */
     async featuresRequestOpts(requestParameters: FeaturesRequest): Promise<runtime.RequestOpts> {
@@ -365,6 +425,53 @@ export class AudioApi extends runtime.BaseAPI {
     }
 
     /**
+     * Creates request options for getJob without sending the request
+     */
+    async getJobRequestOpts(requestParameters: GetJobRequest): Promise<runtime.RequestOpts> {
+        if (requestParameters['jobId'] == null) {
+            throw new runtime.RequiredError(
+                'jobId',
+                'Required parameter "jobId" was null or undefined when calling getJob().'
+            );
+        }
+
+        const queryParameters: any = {};
+
+        const headerParameters: runtime.HTTPHeaders = {};
+
+
+        let urlPath = `/jobs/{job_id}`;
+        urlPath = urlPath.replace('{job_id}', encodeURIComponent(String(requestParameters['jobId'])));
+
+        return {
+            path: urlPath,
+            method: 'GET',
+            headers: headerParameters,
+            query: queryParameters,
+        };
+    }
+
+    /**
+     * Poll an async job. 404 if it doesn\'t exist OR belongs to another tenant (we don\'t leak job existence across accounts).
+     * Get Job
+     */
+    async getJobRaw(requestParameters: GetJobRequest, initOverrides?: RequestInit | runtime.InitOverrideFunction): Promise<runtime.ApiResponse<JobStatus>> {
+        const requestOptions = await this.getJobRequestOpts(requestParameters);
+        const response = await this.request(requestOptions, initOverrides);
+
+        return new runtime.JSONApiResponse(response, (jsonValue) => JobStatusFromJSON(jsonValue));
+    }
+
+    /**
+     * Poll an async job. 404 if it doesn\'t exist OR belongs to another tenant (we don\'t leak job existence across accounts).
+     * Get Job
+     */
+    async getJob(requestParameters: GetJobRequest, initOverrides?: RequestInit | runtime.InitOverrideFunction): Promise<JobStatus> {
+        const response = await this.getJobRaw(requestParameters, initOverrides);
+        return await response.value();
+    }
+
+    /**
      * Creates request options for grade without sending the request
      */
     async gradeRequestOpts(requestParameters: GradeRequest): Promise<runtime.RequestOpts> {

armada/api/clients/nech-api/src/audio/models/JobStatus.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';
+/**
+ * The status of an async job (separate/sources/loops). `result` is populated
+ * once `status == "done"`; `error` once `status == "error"`.
+ * @export
+ * @interface JobStatus
+ */
+export interface JobStatus {
+    /**
+     * 
+     * @type {string}
+     * @memberof JobStatus
+     */
+    jobId: string;
+    /**
+     * 
+     * @type {string}
+     * @memberof JobStatus
+     */
+    type: string;
+    /**
+     * 
+     * @type {string}
+     * @memberof JobStatus
+     */
+    status: string;
+    /**
+     * 
+     * @type {number}
+     * @memberof JobStatus
+     */
+    progress: number;
+    /**
+     * 
+     * @type {{ [key: string]: any; }}
+     * @memberof JobStatus
+     */
+    result?: { [key: string]: any; } | null;
+    /**
+     * 
+     * @type {string}
+     * @memberof JobStatus
+     */
+    resultRef?: string | null;
+    /**
+     * 
+     * @type {string}
+     * @memberof JobStatus
+     */
+    error?: string | null;
+    /**
+     * 
+     * @type {number}
+     * @memberof JobStatus
+     */
+    created: number;
+    /**
+     * 
+     * @type {number}
+     * @memberof JobStatus
+     */
+    started?: number | null;
+    /**
+     * 
+     * @type {number}
+     * @memberof JobStatus
+     */
+    finished?: number | null;
+}
+
+/**
+ * Check if a given object implements the JobStatus interface.
+ */
+export function instanceOfJobStatus(value: object): value is JobStatus {
+    if ((!('jobId' in value) && !('job_id' in value)) || (value['jobId'] === undefined && value['job_id'] === undefined)) return false;
+    if (!('type' in value) || value['type'] === undefined) return false;
+    if (!('status' in value) || value['status'] === undefined) return false;
+    if (!('progress' in value) || value['progress'] === undefined) return false;
+    if (!('created' in value) || value['created'] === undefined) return false;
+    return true;
+}
+
+export function JobStatusFromJSON(json: any): JobStatus {
+    return JobStatusFromJSONTyped(json, false);
+}
+
+export function JobStatusFromJSONTyped(json: any, ignoreDiscriminator: boolean): JobStatus {
+    if (json == null) {
+        return json;
+    }
+    return {
+        
+        'jobId': json['job_id'],
+        'type': json['type'],
+        'status': json['status'],
+        'progress': json['progress'],
+        'result': json['result'] == null ? undefined : json['result'],
+        'resultRef': json['result_ref'] == null ? undefined : json['result_ref'],
+        'error': json['error'] == null ? undefined : json['error'],
+        'created': json['created'],
+        'started': json['started'] == null ? undefined : json['started'],
+        'finished': json['finished'] == null ? undefined : json['finished'],
+    };
+}
+
+export function JobStatusToJSON(json: any): JobStatus {
+    return JobStatusToJSONTyped(json, false);
+}
+
+export function JobStatusToJSONTyped(value?: JobStatus | null, ignoreDiscriminator: boolean = false): any {
+    if (value == null) {
+        return value;
+    }
+
+    return {
+        
+        'job_id': value['jobId'],
+        'type': value['type'],
+        'status': value['status'],
+        'progress': value['progress'],
+        'result': value['result'],
+        'result_ref': value['resultRef'],
+        'error': value['error'],
+        'created': value['created'],
+        'started': value['started'],
+        'finished': value['finished'],
+    };
+}
+

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

 /* tslint:disable */
 /* eslint-disable */
 export * from './HTTPValidationError';
+export * from './JobStatus';
 export * from './LocationInner';
 export * from './ValidationError';

armada/api/clients/nech-api/tsconfig.json

@@ -7,6 +7,10 @@
     "outDir": "dist",
     "rootDir": "src",
     "strict": true,
+    // src/ is 100% generated; typescript-fetch guards index a camel/snake key
+    // union that trips TS7053 under noImplicitAny. The emitted .d.ts is still
+    // fully typed for consumers — this only relaxes the generator's own guards.
+    "noImplicitAny": false,
     "esModuleInterop": true,
     "skipLibCheck": true,
     "forceConsistentCasingInFileNames": true,

armada/api/clients/openapi.json

@@ -45,6 +45,90 @@
         }
       }
     },
+    "/jobs/{job_id}": {
+      "get": {
+        "tags": [
+          "audio"
+        ],
+        "summary": "Get Job",
+        "description": "Poll an async job. 404 if it doesn't exist OR belongs to another tenant\n(we don't leak job existence across accounts).",
+        "operationId": "getJob",
+        "parameters": [
+          {
+            "name": "job_id",
+            "in": "path",
+            "required": true,
+            "schema": {
+              "type": "string",
+              "title": "Job Id"
+            }
+          }
+        ],
+        "responses": {
+          "200": {
+            "description": "Successful Response",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/JobStatus"
+                }
+              }
+            }
+          },
+          "422": {
+            "description": "Validation Error",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/HTTPValidationError"
+                }
+              }
+            }
+          }
+        }
+      },
+      "delete": {
+        "tags": [
+          "audio"
+        ],
+        "summary": "Cancel Job",
+        "description": "Cancel a pending/running job (cooperative — a running handler stops at its\nnext progress checkpoint). Idempotent: cancelling a finished job is a no-op.",
+        "operationId": "cancelJob",
+        "parameters": [
+          {
+            "name": "job_id",
+            "in": "path",
+            "required": true,
+            "schema": {
+              "type": "string",
+              "title": "Job Id"
+            }
+          }
+        ],
+        "responses": {
+          "200": {
+            "description": "Successful Response",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/JobStatus"
+                }
+              }
+            }
+          },
+          "422": {
+            "description": "Validation Error",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/HTTPValidationError"
+                }
+              }
+            }
+          }
+        }
+      }
+    },
     "/analyze/emotion": {
       "post": {
         "tags": [
@@ -713,6 +797,96 @@
         "type": "object",
         "title": "HTTPValidationError"
       },
+      "JobStatus": {
+        "properties": {
+          "job_id": {
+            "type": "string",
+            "title": "Job Id"
+          },
+          "type": {
+            "type": "string",
+            "title": "Type"
+          },
+          "status": {
+            "type": "string",
+            "title": "Status"
+          },
+          "progress": {
+            "type": "number",
+            "title": "Progress"
+          },
+          "result": {
+            "anyOf": [
+              {
+                "additionalProperties": true,
+                "type": "object"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Result"
+          },
+          "result_ref": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Result Ref"
+          },
+          "error": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Error"
+          },
+          "created": {
+            "type": "number",
+            "title": "Created"
+          },
+          "started": {
+            "anyOf": [
+              {
+                "type": "number"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Started"
+          },
+          "finished": {
+            "anyOf": [
+              {
+                "type": "number"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Finished"
+          }
+        },
+        "type": "object",
+        "required": [
+          "job_id",
+          "type",
+          "status",
+          "progress",
+          "created"
+        ],
+        "title": "JobStatus",
+        "description": "The status of an async job (separate/sources/loops). `result` is populated\nonce `status == \"done\"`; `error` once `status == \"error\"`."
+      },
       "ValidationError": {
         "properties": {
           "loc": {

armada/api/db.py

@@ -55,6 +55,33 @@ CREATE TABLE IF NOT EXISTS aliases (
     content_id TEXT NOT NULL,
     created    REAL NOT NULL
 );
+
+-- async jobs (#25): heavy work (sources/separate/loops) runs out-of-band so the
+-- API stays responsive. The API enqueues (pending) + returns a job_id; a single
+-- worker daemon (worker.py) atomically claims FIFO+priority, runs the registered
+-- handler, writes progress + terminal state. Identity is the gateway tenant
+-- (X-Tenant) stored as `account` text — central auth owns the accounts table now.
+-- status: pending | running | done | error | cancelled.
+CREATE TABLE IF NOT EXISTS jobs (
+    id          TEXT PRIMARY KEY,          -- j_<hex>
+    account     TEXT NOT NULL,             -- gateway tenant (X-Tenant); '' for dev
+    type        TEXT NOT NULL,             -- separate | sources | loops | …
+    params_json TEXT NOT NULL DEFAULT '{}',
+    status      TEXT NOT NULL DEFAULT 'pending',
+    priority    INTEGER NOT NULL DEFAULT 0,
+    progress    REAL NOT NULL DEFAULT 0,
+    result_json TEXT,                      -- inline JSON result
+    result_ref  TEXT,                      -- artifact base path (X-Accel-Redirect)
+    error       TEXT,
+    attempts    INTEGER NOT NULL DEFAULT 0,
+    webhook_url TEXT,
+    created     REAL NOT NULL,
+    started     REAL,
+    finished    REAL
+);
+-- the worker's claim query orders by this; keeps the hot 'pending' scan tight.
+CREATE INDEX IF NOT EXISTS idx_jobs_claim   ON jobs(status, priority DESC, created);
+CREATE INDEX IF NOT EXISTS idx_jobs_account ON jobs(account, created);
 """
 
 

armada/api/deploy/douanier-worker.service

+# Douanier job worker (#25) — systemd --user unit.
+#
+# The worker is a SEPARATE long-lived process from the API container: it claims
+# pending jobs and runs the heavy engines (separate/sources/loops). Runs as a
+# systemd --user service with lingering on, so it survives logout and starts at
+# boot (see memory reference_durable_background_jobs — harness/nohup jobs die on
+# teardown; systemd --user is the durable path).
+#
+# Install on the deploy host (the box that runs the audio service):
+#   loginctl enable-linger $USER
+#   cp deploy/douanier-worker.service ~/.config/systemd/user/
+#   systemctl --user daemon-reload
+#   systemctl --user enable --now douanier-worker
+#   journalctl --user -u douanier-worker -f
+#
+# NOTE: the worker needs the SAME env as the API (DOUANIER_DB, DOUANIER_CACHE,
+# engine venv). Point WorkingDirectory + EnvironmentFile at the deployed copy.
+# If the API runs containerized via `nechapi ship`, either run this worker on the
+# host against a shared DB/cache volume, or add it as a second process in the
+# image — a paved-road decision (see SRE.md / #34).
+
+[Unit]
+Description=Douanier audio-API job worker
+After=network.target
+
+[Service]
+Type=simple
+WorkingDirectory=%h/srv/douanier/api
+EnvironmentFile=%h/.config/douanier/douanier.env
+ExecStart=%h/.virtualenvs/douanier/bin/python -m worker
+Restart=on-failure
+RestartSec=3
+# graceful: the worker traps SIGTERM, finishes the current job's bookkeeping, exits
+KillSignal=SIGTERM
+TimeoutStopSec=120
+
+[Install]
+WantedBy=default.target

armada/api/jobs.py

+"""Async jobs — the backbone for heavy work (sources / separate / loops) (#25).
+
+The API enqueues a job (status=pending) and hands back a `job_id`; a single
+worker daemon (worker.py) atomically claims pending jobs FIFO + priority, runs
+the handler registered for the job's `type`, and writes progress + a terminal
+state (done / error). The API never imports engine code — submit stays light,
+and the worker is the only place the heavy deps load.
+
+Design choices (see task #25):
+- SINGLE serialized worker — one job at a time. erable is one box, 2 GB, no GPU;
+  running two demucs separations at once would OOM. Parallelism is a future lane
+  split, not now. The atomic claim is still race-safe for N workers regardless.
+- Identity is the gateway tenant (X-Tenant), stored as `account` text. Ownership
+  (account match) gates poll/cancel; central auth already gated the scope.
+- Idempotency: the submit path checks the cache first and, on a hit, creates the
+  job already `done` (enqueue(..., result=...)) so a repeat returns instantly.
+
+Handlers register by type:
+
+    @jobs.handler("separate")
+    def run_separate(job, progress):
+        progress(0.5)
+        return {"stems": ...}          # -> stored as the job result
+
+A handler returns a JSON-able dict (inline result) and/or sets `job["result_ref"]`
+via the 2-tuple return `(result, ref)`; raising marks the job error (the worker
+applies the attempt cap / requeue policy).
+"""
+from __future__ import annotations
+
+import json
+import secrets
+import time
+from typing import Callable
+
+import db
+
+# job type -> handler(job: dict, progress: Callable[[float], None]) -> dict | (dict, str)
+HANDLERS: dict[str, Callable] = {}
+
+TERMINAL = ("done", "error", "cancelled")
+MAX_ATTEMPTS = 3
+
+
+def handler(job_type: str):
+    """Register a worker handler for a job type. The worker dispatches on `type`."""
+    def deco(fn: Callable) -> Callable:
+        HANDLERS[job_type] = fn
+        return fn
+    return deco
+
+
+def _new_id() -> str:
+    return "j_" + secrets.token_hex(8)
+
+
+def _row_to_public(row) -> dict:
+    """The API-facing view of a job (no internal columns leak meaningfully)."""
+    return {
+        "job_id": row["id"],
+        "type": row["type"],
+        "status": row["status"],
+        "progress": round(row["progress"], 4),
+        "result": json.loads(row["result_json"]) if row["result_json"] else None,
+        "result_ref": row["result_ref"],
+        "error": row["error"],
+        "created": row["created"],
+        "started": row["started"],
+        "finished": row["finished"],
+    }
+
+
+# ── enqueue ─────────────────────────────────────────────────────────────────────
+def enqueue(account: str, job_type: str, params: dict | None = None, *,
+            priority: int = 0, webhook_url: str | None = None,
+            result: dict | None = None) -> dict:
+    """Create a job and return its public view. If `result` is given (the caller
+    already had a cache hit), the job is born `done` — the idempotent fast path."""
+    jid = _new_id()
+    now = time.time()
+    done = result is not None
+    with db.cursor() as c:
+        c.execute(
+            "INSERT INTO jobs(id, account, type, params_json, status, priority, "
+            "progress, result_json, webhook_url, attempts, created, started, finished) "
+            "VALUES(?,?,?,?,?,?,?,?,?,0,?,?,?)",
+            (jid, account, job_type, json.dumps(params or {}),
+             "done" if done else "pending", priority,
+             1.0 if done else 0.0,
+             json.dumps(result) if done else None,
+             webhook_url, now, now if done else None, now if done else None))
+        row = c.execute("SELECT * FROM jobs WHERE id=?", (jid,)).fetchone()
+    return _row_to_public(row)
+
+
+# ── read / ownership ──────────────────────────────────────────────────────────
+def get(job_id: str, account: str | None = None) -> dict | None:
+    """Public view of a job, or None if missing — or owned by another account
+    when `account` is given (callers should treat that as a 404, not a 403, so
+    job existence doesn't leak across tenants)."""
+    with db.cursor() as c:
+        row = c.execute("SELECT * FROM jobs WHERE id=?", (job_id,)).fetchone()
+    if not row:
+        return None
+    if account is not None and row["account"] != account:
+        return None
+    return _row_to_public(row)
+
+
+def cancel(job_id: str, account: str | None = None) -> str | None:
+    """Cancel a pending/running job. Returns the new status, or None if missing/
+    not owned. A running job is cancelled cooperatively — the handler sees it via
+    is_cancelled() between progress steps."""
+    with db.cursor() as c:
+        row = c.execute("SELECT account, status FROM jobs WHERE id=?", (job_id,)).fetchone()
+        if not row or (account is not None and row["account"] != account):
+            return None
+        if row["status"] in TERMINAL:
+            return row["status"]          # already finished — idempotent no-op
+        c.execute("UPDATE jobs SET status='cancelled', finished=? WHERE id=?",
+                  (time.time(), job_id))
+    return "cancelled"
+
+
+def is_cancelled(job_id: str) -> bool:
+    with db.cursor() as c:
+        row = c.execute("SELECT status FROM jobs WHERE id=?", (job_id,)).fetchone()
+    return bool(row) and row["status"] == "cancelled"
+
+
+# ── worker-side lifecycle ───────────────────────────────────────────────────────
+def claim() -> dict | None:
+    """Atomically claim the next pending job → running. Returns the raw row dict,
+    or None if the queue is empty.
+
+    Race-safety: SQLite (WAL) serializes writers, so two workers can't both
+    transition the same row — the second's `WHERE status='pending'` no longer
+    matches. We pick a candidate, then UPDATE guarded by id+status; rowcount 0
+    means someone beat us, so we retry until we win one or the queue drains."""
+    while True:
+        with db.cursor() as c:
+            cand = c.execute(
+                "SELECT id FROM jobs WHERE status='pending' "
+                "ORDER BY priority DESC, created ASC LIMIT 1").fetchone()
+            if not cand:
+                return None
+            now = time.time()
+            cur = c.execute(
+                "UPDATE jobs SET status='running', started=?, attempts=attempts+1 "
+                "WHERE id=? AND status='pending'", (now, cand["id"]))
+            if cur.rowcount == 1:
+                row = c.execute("SELECT * FROM jobs WHERE id=?", (cand["id"],)).fetchone()
+                return dict(row)
+        # lost the race for this candidate — loop and try the next pending one
+
+
+def set_progress(job_id: str, progress: float) -> None:
+    with db.cursor() as c:
+        c.execute("UPDATE jobs SET progress=? WHERE id=? AND status='running'",
+                  (max(0.0, min(1.0, progress)), job_id))
+
+
+def finish(job_id: str, result: dict | None = None, result_ref: str | None = None) -> None:
+    with db.cursor() as c:
+        c.execute(
+            "UPDATE jobs SET status='done', progress=1.0, result_json=?, result_ref=?, "
+            "finished=? WHERE id=? AND status='running'",
+            (json.dumps(result) if result is not None else None, result_ref,
+             time.time(), job_id))
+
+
+def fail(job_id: str, error: str, *, requeue: bool) -> str:
+    """Mark a job failed. If `requeue` and under the attempt cap, send it back to
+    pending for another worker pass; else terminal error. Returns the new status."""
+    with db.cursor() as c:
+        row = c.execute("SELECT attempts FROM jobs WHERE id=?", (job_id,)).fetchone()
+        if requeue and row and row["attempts"] < MAX_ATTEMPTS:
+            c.execute("UPDATE jobs SET status='pending', started=NULL, error=? WHERE id=?",
+                      (error, job_id))
+            return "pending"
+        c.execute("UPDATE jobs SET status='error', error=?, finished=? WHERE id=?",
+                  (error, time.time(), job_id))
+    return "error"
+
+
+def recover() -> int:
+    """Crash recovery, run once at worker boot: any job left 'running' (the worker
+    died mid-flight) goes back to 'pending' if under the attempt cap, else 'error'.
+    Returns how many were requeued."""
+    now = time.time()
+    with db.cursor() as c:
+        requeued = c.execute(
+            "UPDATE jobs SET status='pending', started=NULL "
+            "WHERE status='running' AND attempts < ?", (MAX_ATTEMPTS,)).rowcount
+        c.execute(
+            "UPDATE jobs SET status='error', error='exceeded attempts (crash loop)', "
+            "finished=? WHERE status='running' AND attempts >= ?", (now, MAX_ATTEMPTS))
+    return requeued

armada/api/tests/test_jobs.py

+"""Async job backbone (#25): enqueue → claim → run → poll/cancel, plus the
+race-safety and crash-recovery guarantees the worker relies on."""
+import threading
+
+import pytest
+from fastapi.testclient import TestClient
+
+import app
+import db
+import jobs
+import worker
+
+client = TestClient(app.app)
+GW = {"X-Tenant": "hexa", "X-Scopes": "api:audio:*"}
+OTHER = {"X-Tenant": "someone-else", "X-Scopes": "api:audio:*"}
+
+
+@pytest.fixture(autouse=True)
+def _clean_jobs():
+    """Each test starts with an empty jobs table (shared temp DB across the suite)."""
+    db.init()
+    with db.cursor() as c:
+        c.execute("DELETE FROM jobs")
+    jobs.HANDLERS.pop("_fake", None)
+    yield
+
+
+# ── enqueue / lifecycle ─────────────────────────────────────────────────────────
+def test_enqueue_is_pending_then_claim_runs_it():
+    j = jobs.enqueue("hexa", "_fake", {"x": 1})
+    assert j["status"] == "pending" and j["job_id"].startswith("j_")
+
+    claimed = jobs.claim()
+    assert claimed["id"] == j["job_id"] and claimed["status"] == "running"
+    # queue now empty
+    assert jobs.claim() is None
+
+
+def test_enqueue_with_result_is_born_done_idempotent_fastpath():
+    j = jobs.enqueue("hexa", "_fake", {"x": 1}, result={"answer": 42})
+    assert j["status"] == "done" and j["result"] == {"answer": 42}
+    assert j["progress"] == 1.0
+    # a done job is never claimed by the worker
+    assert jobs.claim() is None
+
+
+def test_finish_and_fail_requeue_then_cap():
+    j = jobs.enqueue("hexa", "_fake")
+    jid = j["job_id"]
+    jobs.claim()
+    jobs.set_progress(jid, 0.5)
+    assert jobs.get(jid)["progress"] == 0.5
+
+    # fail with requeue: attempt 1 → back to pending
+    assert jobs.fail(jid, "boom", requeue=True) == "pending"
+    jobs.claim()                                  # attempt 2
+    assert jobs.fail(jid, "boom", requeue=True) == "pending"
+    jobs.claim()                                  # attempt 3 == MAX_ATTEMPTS
+    assert jobs.fail(jid, "boom", requeue=True) == "error"
+    assert jobs.get(jid)["error"] == "boom"
+
+
+def test_recover_requeues_inflight():
+    j = jobs.enqueue("hexa", "_fake")
+    jobs.claim()                                  # now 'running'
+    assert jobs.get(j["job_id"])["status"] == "running"
+    n = jobs.recover()                            # simulate worker reboot
+    assert n == 1
+    assert jobs.get(j["job_id"])["status"] == "pending"
+
+
+# ── race-safety: a job is claimed by exactly one worker ─────────────────────────
+def test_no_double_claim_under_concurrency():
+    N = 25
+    ids = {jobs.enqueue("hexa", "_fake")["job_id"] for _ in range(N)}
+    claimed: list[str] = []
+    lock = threading.Lock()
+
+    def drain():
+        while True:
+            job = jobs.claim()
+            if job is None:
+                return
+            with lock:
+                claimed.append(job["id"])
+
+    threads = [threading.Thread(target=drain) for _ in range(6)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    assert sorted(claimed) == sorted(ids)         # every job claimed…
+    assert len(claimed) == len(set(claimed)) == N  # …exactly once
+
+
+# ── worker dispatch ─────────────────────────────────────────────────────────────
+def test_worker_runs_handler_to_done():
+    @jobs.handler("_fake")
+    def _run(job, progress):
+        progress(0.3)
+        assert job["params"] == {"k": "v"}        # params parsed for the handler
+        return {"echo": job["params"]}
+
+    jid = jobs.enqueue("hexa", "_fake", {"k": "v"})["job_id"]
+    worker.run_one(jobs.claim())
+    done = jobs.get(jid)
+    assert done["status"] == "done" and done["result"] == {"echo": {"k": "v"}}
+
+
+def test_worker_missing_handler_errors_no_requeue():
+    jid = jobs.enqueue("hexa", "_nohandler")["job_id"]
+    worker.run_one(jobs.claim())
+    assert jobs.get(jid)["status"] == "error"
+
+
+def test_worker_handler_exception_requeues():
+    @jobs.handler("_fake")
+    def _boom(job, progress):
+        raise RuntimeError("nope")
+
+    jid = jobs.enqueue("hexa", "_fake")["job_id"]
+    worker.run_one(jobs.claim())
+    assert jobs.get(jid)["status"] == "pending"   # transient → requeued under cap
+
+
+def test_worker_respects_cancel_midflight():
+    @jobs.handler("_fake")
+    def _cancel_then_progress(job, progress):
+        jobs.cancel(job["id"])                     # cancelled by someone else
+        progress(0.5)                              # next checkpoint raises _Cancelled
+        return {"should": "not reach"}
+
+    jid = jobs.enqueue("hexa", "_fake")["job_id"]
+    worker.run_one(jobs.claim())
+    assert jobs.get(jid)["status"] == "cancelled"
+
+
+# ── HTTP poll / cancel ──────────────────────────────────────────────────────────
+def test_http_poll_and_ownership():
+    jid = jobs.enqueue("hexa", "_fake", result={"ok": True})["job_id"]
+
+    r = client.get(f"/jobs/{jid}", headers=GW)
+    assert r.status_code == 200
+    body = r.json()
+    assert body["status"] == "done" and body["result"] == {"ok": True}
+
+    # another tenant must NOT see it (404, not 403 — no existence leak)
+    assert client.get(f"/jobs/{jid}", headers=OTHER).status_code == 404
+    # unknown id
+    assert client.get("/jobs/j_doesnotexist", headers=GW).status_code == 404
+
+
+def test_http_cancel():
+    jid = jobs.enqueue("hexa", "_fake")["job_id"]
+    r = client.delete(f"/jobs/{jid}", headers=GW)
+    assert r.status_code == 200 and r.json()["status"] == "cancelled"
+    # idempotent
+    assert client.delete(f"/jobs/{jid}", headers=GW).json()["status"] == "cancelled"
+    # other tenant can't cancel
+    jid2 = jobs.enqueue("hexa", "_fake")["job_id"]
+    assert client.delete(f"/jobs/{jid2}", headers=OTHER).status_code == 404

armada/api/worker.py

+#!/usr/bin/env python3
+"""The job worker daemon (#25) — a single serialized loop.
+
+    ~/.virtualenvs/douanier/bin/python -m worker          # or: python worker.py
+
+Boots, recovers crashed in-flight jobs, then loops: claim one pending job → run
+its registered handler → write result / error → optional webhook. One job at a
+time on purpose (erable is 2 GB / no GPU). Runs as a systemd --user unit in prod
+(see VERDACCIO/SRE ops); SIGTERM/SIGINT finish the current job's cleanup then exit.
+
+Engine handlers register themselves by importing their modules here — the worker
+is the ONLY process that loads the heavy deps (demucs, yt-dlp, …). Until #29/#30/
+#31 land there are none, and the worker just idles politely.
+"""
+from __future__ import annotations
+
+import json
+import signal
+import sys
+import time
+import urllib.request
+
+import jobs
+
+POLL_IDLE_SEC = 1.0      # sleep between empty polls
+_STOP = False
+
+# Engine handlers register on import (each module calls @jobs.handler(...)).
+# Wrapped so a not-yet-built engine doesn't stop the worker booting.
+for _mod in ("engines.sources", "engines.separate", "engines.loops"):
+    try:
+        __import__(_mod)
+    except Exception:
+        pass
+
+
+def _on_signal(signum, _frame):
+    global _STOP
+    _STOP = True
+    print(f"[worker] signal {signum} — finishing up, will exit", flush=True)
+
+
+def _webhook(url: str, payload: dict) -> None:
+    try:
+        req = urllib.request.Request(
+            url, data=json.dumps(payload).encode(),
+            headers={"Content-Type": "application/json"}, method="POST")
+        urllib.request.urlopen(req, timeout=10).close()
+    except Exception as e:                      # webhooks are best-effort, never fatal
+        print(f"[worker] webhook to {url} failed: {e}", flush=True)
+
+
+def run_one(job: dict) -> None:
+    """Dispatch a claimed job to its handler and record the outcome."""
+    jid, jtype = job["id"], job["type"]
+    fn = jobs.HANDLERS.get(jtype)
+    if fn is None:
+        jobs.fail(jid, f"no handler for job type '{jtype}'", requeue=False)
+        return
+
+    def progress(p: float) -> None:
+        if jobs.is_cancelled(jid):
+            raise _Cancelled()
+        jobs.set_progress(jid, p)
+
+    try:
+        params = json.loads(job["params_json"] or "{}")
+        out = fn({**job, "params": params}, progress)
+        # handler may return result dict, or (result, result_ref)
+        result, ref = out if isinstance(out, tuple) else (out, None)
+        if jobs.is_cancelled(jid):
+            return                              # cancelled mid-run — leave it cancelled
+        jobs.finish(jid, result=result, result_ref=ref)
+        status, payload = "done", {"job_id": jid, "status": "done"}
+    except _Cancelled:
+        print(f"[worker] {jid} cancelled mid-flight", flush=True)
+        return
+    except Exception as e:                      # transient — requeue under the cap
+        status = jobs.fail(jid, f"{type(e).__name__}: {e}", requeue=True)
+        payload = {"job_id": jid, "status": status, "error": str(e)}
+        print(f"[worker] {jid} failed ({status}): {e}", flush=True)
+
+    if status in jobs.TERMINAL and job.get("webhook_url"):
+        _webhook(job["webhook_url"], payload)
+
+
+class _Cancelled(Exception):
+    pass
+
+
+def main() -> int:
+    signal.signal(signal.SIGTERM, _on_signal)
+    signal.signal(signal.SIGINT, _on_signal)
+    n = jobs.recover()
+    print(f"[worker] up. recovered {n} in-flight job(s). handlers: "
+          f"{sorted(jobs.HANDLERS) or '(none yet)'}", flush=True)
+    while not _STOP:
+        job = jobs.claim()
+        if job is None:
+            time.sleep(POLL_IDLE_SEC)
+            continue
+        print(f"[worker] running {job['id']} ({job['type']})", flush=True)
+        run_one(job)
+    print("[worker] stopped.", flush=True)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())