CI: push to ipcr-push-85-17-246-67.ip.nsl.sh (the box's cert no longer covers nip.io) [skip ci]

.gitattributes

@@ -0,0 +1,6 @@
+# FIX CRLF always when developer has not set
+# Linux/Mac: git config --global core.autocrlf input
+# Windows: git config --global core.autocrlf true
+# Auto detect text files and perform LF normalization
+* text=auto
+* eol=lf

.gitea/workflows/docker.yml

@@ -0,0 +1,85 @@
+# Gitea twin of .github/workflows/docker.yml. Gitea runs only .gitea/workflows/ once this
+# folder exists, and GitHub never reads it.
+#
+# Publishes to IPCR on watch.nsl.sh instead of ghcr.io:
+#   push  ipcr-push-85-17-246-67.ip.nsl.sh/metadec.eth/<repo>:<tag>   docker login as metadec.eth with the
+#         IPCR_PUSH_TOKEN secret (the metadec.eth namespace token from the IPCR dashboard)
+#   pull  ipcr-watch.nsl.sh/ipns/metadec.eth/metadec.eth/<repo>:<tag>   anonymous
+# The push goes to ipcr-push-85-17-246-67.ip.nsl.sh, the box's own address, not ipcr-push-watch.nsl.sh: Cloudflare fronts
+# *.watch.nsl.sh and refuses any request over 100 MB (413), which a large image layer is.
+# That host's certificate is signed by the box's private Mesh Router CA, given to buildkit from
+# the IPCR_CA_PEM repo variable. docker login would make the job's Docker daemon check the
+# certificate without that CA, so the login is written straight into ~/.docker/config.json.
+# The runner's job image (node:22-bookworm) has no docker CLI, so the static one is
+# installed first; buildx installs itself. No `type=gha` cache: the runner has none.
+
+name: Build and Push Docker Image
+
+on:
+  push:
+    branches: [main, master]
+    tags: ['v*']
+  pull_request:
+    branches: [main, master]
+
+env:
+  REGISTRY: ipcr-push-85-17-246-67.ip.nsl.sh
+  IMAGE_NAME: metadec.eth/${{ github.event.repository.name }}
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      packages: write
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Install Docker CLI
+        run: curl -fsSL https://download.docker.com/linux/static/stable/x86_64/docker-28.5.1.tgz | tar -xz -C /usr/local/bin --strip-components=1 docker/docker
+
+      - name: Trust the Mesh Router CA
+        run: printf '%s\n' "$IPCR_CA_PEM" > /tmp/ipcr-ca.pem
+        env:
+          IPCR_CA_PEM: ${{ vars.IPCR_CA_PEM }}
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+        with:
+          version: latest
+          buildkitd-config-inline: |
+            [registry."${{ env.REGISTRY }}"]
+              ca = ["/tmp/ipcr-ca.pem"]
+
+      - name: Log in to IPCR
+        if: github.event_name != 'pull_request'
+        env:
+          IPCR_PUSH_TOKEN: ${{ secrets.IPCR_PUSH_TOKEN }}
+        run: |
+          mkdir -p ~/.docker
+          printf '{"auths":{"%s":{"auth":"%s"}}}\n' "$REGISTRY" "$(printf 'metadec.eth:%s' "$IPCR_PUSH_TOKEN" | base64 -w0)" > ~/.docker/config.json
+
+      - name: Extract metadata
+        id: meta
+        uses: docker/metadata-action@v5
+        with:
+          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
+          tags: |
+            type=ref,event=branch
+            type=ref,event=pr
+            type=semver,pattern={{version}}
+            # A fix to release 1.2.3 is tagged v1.2.3.1: four numbers are not semver, so
+            # the pattern above makes no tag for it; this one does (image tag 1.2.3.1).
+            type=match,pattern=v(\d+\.\d+\.\d+\.\d+),group=1
+            type=semver,pattern={{major}}.{{minor}}
+            type=sha
+
+      - name: Build and push
+        uses: docker/build-push-action@v5
+        with:
+          context: .
+          push: ${{ github.event_name != 'pull_request' }}
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}

.github/workflows/docker.yml

@@ -42,6 +42,9 @@ jobs:
             type=ref,event=branch
             type=ref,event=pr
             type=semver,pattern={{version}}
+            # A fix to release 1.2.3 is tagged v1.2.3.1: four numbers are not semver, so
+            # the pattern above makes no tag for it; this one does (image tag 1.2.3.1).
+            type=match,pattern=v(\d+\.\d+\.\d+\.\d+),group=1
             type=semver,pattern={{major}}.{{minor}}
             type=sha
 

README.md

@@ -38,14 +38,28 @@ in `src/plugin.ts`.
 
 ## How a frame is chosen
 
-1. Candidate offsets at 20 / 45 / 70 % of `fileinfo/duration`, clamped 2s away
-   from both ends (the head is logos and black, the tail is credits).
-2. At each offset, one ffmpeg run with **input seeking** (`-ss` before `-i`, so
-   it Range-seeks over WebDAV instead of decoding from zero) and
-   `thumbnail=60`, which picks the most representative of the next 60 frames.
-3. The encoded JPEG is measured (mean + stddev of a 32×32 grayscale reduction).
+1. **Record first, file second.** `fileinfo/duration` and the stream table
+   from the record; if either is missing, one header-only `ffprobe` reads it
+   from the file (a record with no stream data at all has been seen on the dev
+   stack). meta-sort treats a `failed` or
+   `skipped` ffmpeg dependency as complete, so a missing duration is a normal
+   case, not a bug. Only if the probe fails too does it fall back to a blind
+   10s then 0s — and that fallback lands on distributor idents and opening
+   credits, so it is a last resort.
+2. Candidate offsets at 20 / 45 / 70 % of the duration, clamped 2s away from
+   both ends (the head is logos and black, the tail is credits).
+3. At each offset, one ffmpeg run with **input seeking** (`-ss` before `-i`, so
+   it Range-seeks over WebDAV instead of decoding from zero), scaled down
+   **before** `thumbnail=60` picks the most representative of the next 60
+   frames. That order matters: `thumbnail` buffers all 60 frames, and at full
+   1080p a single grab peaked at 409 MiB versus 196 MiB scaled-first.
+4. The encoded JPEG is measured (mean + stddev of a 32×32 grayscale reduction).
    Black, blown-out and flat frames are rejected and the next offset is tried.
-4. All candidates rejected → `skipped`, no field written.
+5. All candidates rejected → `skipped`, no field written.
+
+Extractions run **one at a time** by default (`maxConcurrent`). meta-sort's
+background queue sends several tasks at once, and four concurrent 1080p grabs
+OOM-killed the 512 MiB container; with one CPU, parallel decodes buy nothing.
 
 The primary video stream is the largest one that **looks like moving video** —
 not `0:v:0`, and not simply the largest. ffprobe reports attached cover art as
@@ -68,6 +82,7 @@ curl -k -X PUT https://metasort-dev.localhost:8180/api/plugins/still-extractor/c
 | `maxWidth` | `640` | Frame is scaled down to at most this width |
 | `jpegQuality` | `4` | ffmpeg `-q:v`; lower is better quality |
 | `frameTimeoutMs` | `120000` | Hard kill for one ffmpeg grab |
+| `maxConcurrent` | `1` | Extractions allowed to run at once in this instance |
 
 ## Development
 
@@ -80,3 +95,18 @@ docker build -t metamesh-plugin-still-extractor:main .
 
 Depends on `file-info` (for `fileType`) and `ffmpeg` (for `fileinfo/duration`
 and the `stream/{n}` table). Runs on the **background** queue.
+
+### What meta-sort actually sends
+
+⚠ The `existingMeta` in a `/process` request is the **nested document form** —
+`fileinfo: { duration }`, `stream: [...]`, `cids: [...]` — not the flat
+`fileinfo/duration` / `stream/{n}` keys you see in meta-core's flat record.
+Reading only the flat key made every task fall back to 10s/0s on the dev stack.
+`durationFromMeta` and `selectPrimaryVideoStream` accept both shapes; keep it
+that way.
+
+A replay or rescan queues several tasks for the same file. Each reads the
+record at dispatch, so the plugin re-checks `still` in meta-core once it holds
+an extraction slot, instead of recomputing the same frame for every duplicate.
+A file that yields no usable frame is remembered in memory for an hour, so its
+queued duplicates skip instead of decoding it again; a later rescan retries.

src/beacon-advertise.ts

@@ -0,0 +1,188 @@
+/**
+ * Beacon v2 advertiser — makes this plugin findable by meta-sort's Plugins
+ * page (Scan → Add) without anyone typing its URL.
+ *
+ * Advertises one resource, cap `metamesh.enrich/<manifest id>`, on the beacon
+ * group (239.255.99.1:9099): unsolicited every 10 s, and unicast in reply to
+ * any probe whose `want` matches. Says `bye` on shutdown. It keeps no view of
+ * other nodes — a plugin only needs to be found. Spec:
+ * meta-root docs/project-architecture/beacon-v2.md.
+ *
+ * Env:
+ *   ENABLE_UDP_DISCOVERY  false/0/no/off disables
+ *   BEACON_ADVERTISE_URL  endpoints.http (default http://$HOSTNAME:$PORT)
+ *   BEACON_BINDS          the one meta-sort instance allowed to list it
+ *                         (meta-sort sets this on containers it spawns)
+ *   BEACON_GROUP / BEACON_PORT / BEACON_INTERVAL_MS
+ *
+ * ⚠ MIRRORED FILE. Byte-identical copies live in every TypeScript
+ * metamesh-plugin-*; see meta-root `scripts/check-mirrors.sh`. Never fatal: a
+ * host that blocks multicast logs a warning and the plugin serves on.
+ */
+
+import dgram from 'dgram';
+import { hostname, networkInterfaces } from 'os';
+
+const PROTO = 'beacon';
+const VERSION = 2;
+
+interface Resource {
+    id: string;
+    caps: string[];
+    endpoints: Record<string, string>;
+    binds?: string;
+}
+
+interface NodeInfo {
+    name: string;
+    instance: string;
+    version?: string;
+}
+
+function envNum(key: string): number | undefined {
+    const n = Number(process.env[key]);
+    return Number.isFinite(n) && n > 0 ? n : undefined;
+}
+
+function enabled(): boolean {
+    const v = (process.env.ENABLE_UDP_DISCOVERY ?? '').trim().toLowerCase();
+    return !['false', '0', 'no', 'off'].includes(v);
+}
+
+/** IPv4 addresses of every up, non-internal interface. */
+function interfaceAddresses(): string[] {
+    const out: string[] = [];
+    for (const list of Object.values(networkInterfaces())) {
+        for (const ni of list ?? []) {
+            const v4 = (ni as { family: string | number }).family === 'IPv4' || (ni as { family: string | number }).family === 4;
+            if (v4 && !ni.internal && ni.address) out.push(ni.address);
+        }
+    }
+    return out;
+}
+
+/** Same matching rules as the spec: `*`, `x/*`, optional `@contract`. */
+function capMatches(pattern: string, cap: string): boolean {
+    if (pattern === '*') return true;
+    const split = (s: string): [string, string | null] => {
+        const i = s.lastIndexOf('@');
+        return i >= 0 ? [s.slice(0, i), s.slice(i + 1)] : [s, null];
+    };
+    const [pb, pc] = split(pattern);
+    const [cb, cc] = split(cap);
+    if (pc !== null && cc !== pc) return false;
+    if (pb.endsWith('/*')) {
+        const prefix = pb.slice(0, -2);
+        const rest = cb.startsWith(prefix) ? cb.slice(prefix.length) : '';
+        return rest.startsWith('/') && rest.length > 1;
+    }
+    return pb === cb;
+}
+
+let socket: dgram.Socket | null = null;
+let timer: NodeJS.Timeout | null = null;
+let node: NodeInfo | null = null;
+let resource: Resource | null = null;
+let ifaces: string[] = [];
+let group = '239.255.99.1';
+let udpPort = 9099;
+
+function send(msg: object, dest?: { address: string; port: number }): void {
+    const s = socket;
+    if (!s) return;
+    const body = Buffer.from(JSON.stringify(msg));
+    if (dest) {
+        s.send(body, dest.port, dest.address, () => {});
+        return;
+    }
+    // Once per interface: the default route alone reaches one network.
+    for (const addr of ifaces) {
+        try {
+            s.setMulticastInterface(addr);
+            s.send(body, udpPort, group, () => {});
+        } catch (e) {
+            console.warn(`[beacon] send on ${addr} failed:`, e);
+        }
+    }
+}
+
+function advertise(dest?: { address: string; port: number }): void {
+    if (!node || !resource) return;
+    send({ proto: PROTO, v: VERSION, type: 'advertise', node, resources: [resource] }, dest);
+}
+
+/**
+ * Start advertising this plugin. Call once the HTTP server is listening, so a
+ * consumer that hears it can reach it. Idempotent.
+ */
+export function startBeacon(pluginId: string, version: string, httpPort: number): void {
+    if (socket || !enabled()) return;
+    group = process.env.BEACON_GROUP || group;
+    udpPort = envNum('BEACON_PORT') ?? udpPort;
+    const intervalMs = envNum('BEACON_INTERVAL_MS') ?? 10_000;
+    const instance = process.env.HOSTNAME || hostname();
+    const url = (process.env.BEACON_ADVERTISE_URL || `http://${instance}:${httpPort}`).replace(/\/+$/, '');
+    const binds = (process.env.BEACON_BINDS || '').trim();
+
+    node = { name: `metamesh-plugin-${pluginId}`, instance, ...(version ? { version } : {}) };
+    resource = {
+        id: pluginId,
+        caps: [`metamesh.enrich/${pluginId}`],
+        endpoints: { http: url, manifest: '/manifest' },
+        ...(binds ? { binds } : {}),
+    };
+
+    const s = dgram.createSocket({ type: 'udp4', reuseAddr: true });
+    socket = s;
+    s.on('error', (err) => console.warn('[beacon] socket error:', err.message));
+    s.on('message', (buf, rinfo) => {
+        let m: { proto?: string; v?: number; type?: string; from?: string; want?: string[] };
+        try {
+            m = JSON.parse(buf.toString('utf8'));
+        } catch {
+            return; // foreign traffic on the shared group
+        }
+        if (m?.proto !== PROTO || m.v !== VERSION || m.type !== 'probe' || m.from === instance) return;
+        const want = Array.isArray(m.want) ? m.want : [];
+        if (want.length === 0 || resource!.caps.some((c) => want.some((p) => capMatches(p, c)))) {
+            advertise({ address: rinfo.address, port: rinfo.port });
+        }
+    });
+    s.bind(udpPort, '0.0.0.0', () => {
+        try {
+            s.setMulticastTTL(1);
+        } catch {
+            /* best effort */
+        }
+        ifaces = interfaceAddresses();
+        for (const addr of ifaces) {
+            try {
+                s.addMembership(group, addr);
+            } catch (e) {
+                console.warn(`[beacon] join ${group} on ${addr} failed:`, e);
+            }
+        }
+        console.log(`[beacon] advertising metamesh.enrich/${pluginId} at ${url} on ${group}:${udpPort}`);
+        advertise();
+        timer = setInterval(() => advertise(), intervalMs);
+        timer.unref?.();
+    });
+}
+
+/** Say `bye` so consumers drop this plugin at once. */
+export async function stopBeacon(): Promise<void> {
+    if (!socket || !node) return;
+    if (timer) clearInterval(timer);
+    send({ proto: PROTO, v: VERSION, type: 'bye', node });
+    // Let the bye datagrams leave before the socket closes under them.
+    await new Promise((r) => setTimeout(r, 50));
+    const s = socket;
+    socket = null;
+    await new Promise<void>((resolve) => {
+        try {
+            s.close(() => resolve());
+        } catch {
+            resolve();
+        }
+    });
+}

src/index.ts

@@ -19,6 +19,7 @@
 import Fastify from 'fastify';
 import type { HealthResponse, ProcessRequest, ProcessResponse, CallbackPayload, ConfigureRequest, ConfigureResponse } from './types.js';
 import { manifest, process as processFile, configure } from './plugin.js';
+import { startBeacon, stopBeacon } from './beacon-advertise.js';
 
 const app = Fastify({ logger: true });
 let ready = false;
@@ -70,11 +71,13 @@ app.post<{ Body: ProcessRequest }>('/process', async (request, reply) => {
 const port = parseInt(process.env.PORT || '8080', 10);
 
 app.listen({ port, host: '0.0.0.0' }).then(() => {
+    startBeacon(manifest.id, manifest.version, port);
     ready = true;
     console.log(`[${manifest.id}] Listening on port ${port}`);
 });
 
 process.on('SIGTERM', async () => {
+    await stopBeacon();
     ready = false;
     await app.close();
     process.exit(0);

src/meta-core-client.ts

@@ -1,93 +1,380 @@
 /**
- * meta-core API Client for writing metadata
- * Fails gracefully if meta-core is unavailable (logs warning, returns without throwing)
+ * meta-core client: signed claim writes (MMUR2) + plain reads.
+ *
+ * MIRRORED byte-for-byte across every `metamesh-plugin-*` TypeScript plugin and
+ * meta-sort-core (`src/kv/meta-core-client.ts`) — guarded by
+ * `scripts/check-mirrors.sh`. Edit one copy, then copy it everywhere.
+ *
+ * Every write is a vote signed by THIS container's key
+ * (docs/implementation/metacore-claims-model.md §3, §9):
+ *
+ *   bytes_to_sign = "MMUR2\n" || canonical CBOR [cid, uid, key, value, ts]
+ *   sig           = ECDSA-secp256k1(SHA-256(bytes)), compact r||s, low-S
+ *
+ * and is posted to `POST {baseUrl}/api/claims` — meta-core directly, or
+ * meta-sort's pass-through relay when the plugin runs under meta-sort. The
+ * encoding is pinned by meta-core's `internal/claim/testdata/claim-vectors.json`
+ * (vendored as `test/claim-vectors.json`).
+ *
+ * Key: `METAMESH_SIGNING_KEY` (64 hex chars) or `METAMESH_SIGNING_KEY_FILE`
+ * (read; generated and written 0600 if missing). With neither, writes are
+ * refused (logged, return false) — no silent throwaway identities.
+ *
+ * Fails gracefully if meta-core is unavailable: logs and returns, never throws.
  */
 
+import * as crypto from 'crypto';
+import * as fs from 'fs';
+import * as path from 'path';
+
+// ---------------------------------------------------------------- encoding --
+
+const DOMAIN = Buffer.from('MMUR2\n', 'utf8');
+const B58 = '123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz';
+const N = BigInt('0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141');
+const HALF_N = N >> 1n;
+
+/** A claim value: text (public), bytes (private envelope) or null (retraction). */
+export type ClaimValue = string | Uint8Array | null;
+
+export interface SignedClaim {
+    cid: string;
+    uid: string;
+    key: string;
+    value: ClaimValue;
+    ts: number;
+    sig: Uint8Array;
+}
+
+function head(major: number, n: number | bigint): Buffer {
+    const m = major << 5;
+    const v = BigInt(n);
+    if (v < 24n) return Buffer.from([m | Number(v)]);
+    if (v <= 0xffn) return Buffer.from([m | 24, Number(v)]);
+    if (v <= 0xffffn) {
+        const b = Buffer.alloc(3);
+        b[0] = m | 25;
+        b.writeUInt16BE(Number(v), 1);
+        return b;
+    }
+    if (v <= 0xffffffffn) {
+        const b = Buffer.alloc(5);
+        b[0] = m | 26;
+        b.writeUInt32BE(Number(v), 1);
+        return b;
+    }
+    const b = Buffer.alloc(9);
+    b[0] = m | 27;
+    b.writeBigUInt64BE(v, 1);
+    return b;
+}
+
+function text(s: string): Buffer {
+    const raw = Buffer.from(s, 'utf8');
+    return Buffer.concat([head(3, raw.length), raw]);
+}
+
+/** The exact bytes an MMUR2 signature covers. */
+export function signedBytes(cid: string, uid: string, key: string, value: ClaimValue, ts: number): Buffer {
+    let v: Buffer;
+    if (value === null) v = Buffer.from([0xf6]);
+    else if (typeof value === 'string') v = text(value);
+    else v = Buffer.concat([head(2, value.length), Buffer.from(value)]);
+    return Buffer.concat([DOMAIN, head(4, 5), text(cid), text(uid), text(key), v, head(0, ts)]);
+}
+
+function base58Encode(bytes: Uint8Array): string {
+    let n = BigInt('0x' + (Buffer.from(bytes).toString('hex') || '0'));
+    let out = '';
+    while (n > 0n) {
+        out = B58[Number(n % 58n)] + out;
+        n /= 58n;
+    }
+    for (const b of bytes) {
+        if (b !== 0) break;
+        out = '1' + out;
+    }
+    return out;
+}
+
+function base58Decode(s: string): Buffer {
+    let n = 0n;
+    for (const ch of s) {
+        const i = B58.indexOf(ch);
+        if (i < 0) throw new Error(`invalid base58 character ${ch}`);
+        n = n * 58n + BigInt(i);
+    }
+    let hex = n === 0n ? '' : n.toString(16);
+    if (hex.length % 2) hex = '0' + hex;
+    let zeros = 0;
+    while (zeros < s.length && s[zeros] === '1') zeros++;
+    return Buffer.concat([Buffer.alloc(zeros), Buffer.from(hex, 'hex')]);
+}
+
+function b64url(b: Buffer): string {
+    return b.toString('base64url');
+}
+
+function publicKeyFromCompressed(compressed: Buffer): crypto.KeyObject {
+    const full = crypto.ECDH.convertKey(compressed, 'secp256k1', undefined, undefined, 'uncompressed') as Buffer;
+    return crypto.createPublicKey({
+        key: { kty: 'EC', crv: 'secp256k1', x: b64url(full.subarray(1, 33)), y: b64url(full.subarray(33, 65)) },
+        format: 'jwk',
+    });
+}
+
+/** Rewrite a compact signature to its low-S form (MMUR2 refuses high-S). */
+function lowS(sig: Buffer): Buffer {
+    const s = BigInt('0x' + sig.subarray(32).toString('hex'));
+    if (s <= HALF_N) return sig;
+    const fixed = Buffer.from((N - s).toString(16).padStart(64, '0'), 'hex');
+    return Buffer.concat([sig.subarray(0, 32), fixed]);
+}
+
+/** True when the claim's signature is a valid low-S signature by its uid. */
+export function verifyClaim(c: SignedClaim): boolean {
+    try {
+        const sig = Buffer.from(c.sig);
+        if (sig.length !== 64) return false;
+        if (BigInt('0x' + sig.subarray(32).toString('hex')) > HALF_N) return false;
+        if (!c.uid.startsWith('z')) return false;
+        const pub = base58Decode(c.uid.slice(1));
+        if (pub.length !== 33) return false;
+        return crypto.verify(
+            'sha256',
+            signedBytes(c.cid, c.uid, c.key, c.value, c.ts),
+            { key: publicKeyFromCompressed(pub), dsaEncoding: 'ieee-p1363' },
+            sig,
+        );
+    } catch {
+        return false;
+    }
+}
+
+// ----------------------------------------------------------------- signer --
+
+export class ClaimSigner {
+    readonly uid: string;
+    private readonly key: crypto.KeyObject;
+    private lastTs = 0;
+
+    constructor(privateKeyHex: string) {
+        const hex = privateKeyHex.trim().replace(/^0x/i, '');
+        if (!/^[0-9a-fA-F]{64}$/.test(hex)) throw new Error('signing key must be 64 hex characters');
+        const d = Buffer.from(hex, 'hex');
+        const ecdh = crypto.createECDH('secp256k1');
+        ecdh.setPrivateKey(d);
+        const full = ecdh.getPublicKey(null, 'uncompressed');
+        this.uid = 'z' + base58Encode(ecdh.getPublicKey(null, 'compressed'));
+        this.key = crypto.createPrivateKey({
+            key: {
+                kty: 'EC', crv: 'secp256k1', d: b64url(d),
+                x: b64url(full.subarray(1, 33)), y: b64url(full.subarray(33, 65)),
+            },
+            format: 'jwk',
+        });
+    }
+
+    /** Milliseconds, strictly increasing within this process (ts is the version). */
+    nextTs(): number {
+        this.lastTs = Math.max(Date.now(), this.lastTs + 1);
+        return this.lastTs;
+    }
+
+    sign(cid: string, key: string, value: ClaimValue, ts: number = this.nextTs()): SignedClaim {
+        const raw = crypto.sign('sha256', signedBytes(cid, this.uid, key, value, ts), {
+            key: this.key,
+            dsaEncoding: 'ieee-p1363',
+        });
+        return { cid, uid: this.uid, key, value, ts, sig: lowS(raw) };
+    }
+}
+
+let cachedSigner: ClaimSigner | null | undefined;
+
+/**
+ * The process signer from the environment, or null when no key is configured.
+ * Read lazily on first write; `resetSignerForTests()` forgets it.
+ */
+export function loadSigner(): ClaimSigner | null {
+    if (cachedSigner !== undefined) return cachedSigner;
+    const inline = process.env.METAMESH_SIGNING_KEY;
+    const file = process.env.METAMESH_SIGNING_KEY_FILE;
+    try {
+        if (inline) {
+            cachedSigner = new ClaimSigner(inline);
+        } else if (file) {
+            if (!fs.existsSync(file)) {
+                fs.mkdirSync(path.dirname(file), { recursive: true });
+                fs.writeFileSync(file, crypto.randomBytes(32).toString('hex') + '\n', { mode: 0o600 });
+            }
+            cachedSigner = new ClaimSigner(fs.readFileSync(file, 'utf8'));
+        } else {
+            cachedSigner = null;
+        }
+    } catch (error) {
+        console.warn(`[MetaCoreClient] Signing key unusable: ${error instanceof Error ? error.message : String(error)}`);
+        cachedSigner = null;
+    }
+    if (cachedSigner) console.log(`[MetaCoreClient] Signing claims as ${cachedSigner.uid}`);
+    else console.warn('[MetaCoreClient] No METAMESH_SIGNING_KEY / METAMESH_SIGNING_KEY_FILE — metadata writes are refused');
+    return cachedSigner;
+}
+
+export function resetSignerForTests(): void {
+    cachedSigner = undefined;
+}
+
+/** The §3.4 JSON form. */
+export function claimToJSON(c: SignedClaim): Record<string, unknown> {
+    let value: unknown = null;
+    if (typeof c.value === 'string') value = c.value;
+    else if (c.value) value = { b64: Buffer.from(c.value).toString('base64') };
+    return { cid: c.cid, uid: c.uid, key: c.key, value, ts: c.ts, sig: Buffer.from(c.sig).toString('base64') };
+}
+
+// ----------------------------------------------------------------- client --
+
 export class MetaCoreClient {
     constructor(private baseUrl: string) {}
 
     private async safeFetch(url: string, options: RequestInit): Promise<Response | null> {
         try {
-            return await fetch(url, { ...options, signal: AbortSignal.timeout(5000) });
+            return await fetch(url, { ...options, signal: AbortSignal.timeout(10000) });
         } catch (error) {
-            console.warn(`[MetaCoreClient] Warning: meta-core unavailable at ${this.baseUrl}`);
+            console.warn(`[MetaCoreClient] Warning: meta-core unavailable at ${this.baseUrl}: ${error instanceof Error ? error.message : String(error)}`);
             return null;
         }
     }
 
-    async setProperty(hashId: string, key: string, value: string): Promise<void> {
-        const response = await this.safeFetch(`${this.baseUrl}/meta/${hashId}/${key}`, {
-            method: 'PUT',
+    /**
+     * Sign and post claims about `cid`. True when every claim was accepted (or
+     * was already stored: `unchanged` / `stale` are not failures).
+     */
+    async writeClaims(cid: string, values: Array<[string, ClaimValue]>): Promise<boolean> {
+        if (values.length === 0) return true;
+        const signer = loadSigner();
+        if (!signer) return false;
+        const claims = values.map(([key, value]) => claimToJSON(signer.sign(cid, key, value)));
+        const response = await this.safeFetch(`${this.baseUrl}/api/claims`, {
+            method: 'POST',
             headers: { 'Content-Type': 'application/json' },
-            body: JSON.stringify({ value }),
+            body: JSON.stringify(claims),
         });
-        if (response && !response.ok) {
-            console.warn(`[MetaCoreClient] Failed to set property: ${response.status}`);
+        if (!response) return false;
+        if (!response.ok) {
+            console.warn(`[MetaCoreClient] Claims for ${cid} rejected: ${response.status}`);
+            return false;
         }
+        const statuses = await response.json().catch(() => null) as Array<{ status?: string; reason?: string }> | null;
+        const bad = (statuses ?? []).filter((s) => s.status === 'bad_signature' || s.status === 'bad_request');
+        if (bad.length > 0) {
+            console.warn(`[MetaCoreClient] ${bad.length} claim(s) for ${cid} refused: ${bad[0].status} ${bad[0].reason ?? ''}`);
+            return false;
+        }
+        return true;
+    }
+
+    /** One text claim. */
+    async setProperty(hashId: string, key: string, value: string): Promise<boolean> {
+        return this.writeClaims(hashId, [[key, String(value)]]);
+    }
+
+    /** One text claim per entry, in a single batch. */
+    async mergeMetadata(hashId: string, metadata: Record<string, string>): Promise<boolean> {
+        return this.writeClaims(hashId, Object.entries(metadata).map(([k, v]) => [k, String(v)]));
+    }
+
+    /**
+     * Retract this signer's own vote for `key`. Other signers' votes for the
+     * same key are untouched (there are no negative votes).
+     */
+    async deleteProperty(hashId: string, key: string): Promise<boolean> {
+        return this.writeClaims(hashId, [[key, null]]);
+    }
+
+    /**
+     * Append a member to a legacy csv-set field (`genres`, `studio`, `tags`,
+     * `announceList`, …) — the comma-joined shape readers still parse (the
+     * METADATA_KEYS §14.12 key-set migration is out of scope for phase 1).
+     *
+     * Under claims a signer owns only its OWN vote, so this reads this
+     * signer's current claim for (cid, key) from
+     * `GET /api/meta/{cid}/claims?key=<key>` (other signers' claims are
+     * ignored), adds the member if absent and signs the new full value as one
+     * text claim. Members are kept raw (no escaping), same as the old
+     * meta-sort `_add` proxy. A member already present writes nothing.
+     */
+    async addToSet(hashId: string, key: string, value: string): Promise<boolean> {
+        const member = String(value);
+        if (!member) return true;
+        const signer = loadSigner();
+        if (!signer) return false;
+        const current = await this.ownClaimValue(hashId, key, signer.uid);
+        if (current === undefined) return false; // could not read: never clobber our own set
+        const members = current ? current.split(',').map((v) => v.trim()).filter(Boolean) : [];
+        if (members.includes(member)) return true;
+        return this.writeClaims(hashId, [[key, [...members, member].join(',')]]);
+    }
+
+    /**
+     * This signer's newest text claim for (cid, key): the value, `null` when it
+     * has none, `undefined` when the claims could not be read.
+     */
+    private async ownClaimValue(cid: string, key: string, uid: string): Promise<string | null | undefined> {
+        const url = `${this.baseUrl}/api/meta/${encodeURIComponent(cid)}/claims?key=${encodeURIComponent(key)}`;
+        const response = await this.safeFetch(url, { method: 'GET' });
+        if (!response) return undefined;
+        if (response.status === 404) return null;
+        if (!response.ok) {
+            console.warn(`[MetaCoreClient] Failed to read claims for ${cid}/${key}: ${response.status}`);
+            return undefined;
+        }
+        const body = await response.json().catch(() => null) as unknown;
+        const list = (Array.isArray(body) ? body : (body as { claims?: unknown })?.claims) as
+            Array<{ uid?: string; key?: string; value?: unknown; ts?: number }> | undefined;
+        let best: { value: string; ts: number } | null = null;
+        for (const c of Array.isArray(list) ? list : []) {
+            if (c.uid !== uid || c.key !== key || typeof c.value !== 'string') continue;
+            const ts = Number(c.ts) || 0;
+            if (!best || ts > best.ts) best = { value: c.value, ts };
+        }
+        return best ? best.value : null;
     }
 
     async getProperty(hashId: string, key: string): Promise<string | null> {
         const response = await this.safeFetch(`${this.baseUrl}/meta/${hashId}/${key}`, { method: 'GET' });
-        if (!response || response.status === 404) return null;
-        if (!response.ok) return null;
-        const data = await response.json() as { value?: string };
-        return data.value ?? null;
+        if (!response || !response.ok) return null;
+        const data = await response.json() as { value?: unknown };
+        return data.value === undefined || data.value === null ? null : String(data.value);
     }
 
-    async mergeMetadata(hashId: string, metadata: Record<string, string>): Promise<void> {
-        const response = await this.safeFetch(`${this.baseUrl}/meta/${hashId}`, {
-            method: 'PATCH',
-            headers: { 'Content-Type': 'application/json' },
-            body: JSON.stringify(metadata),
-        });
-        if (response && !response.ok) {
-            console.warn(`[MetaCoreClient] Failed to merge metadata: ${response.status}`);
-        }
-    }
-
-    async deleteProperty(hashId: string, key: string): Promise<void> {
-        const response = await this.safeFetch(`${this.baseUrl}/meta/${hashId}/${key}`, { method: 'DELETE' });
-        if (response && !response.ok && response.status !== 404) {
-            console.warn(`[MetaCoreClient] Failed to delete property: ${response.status}`);
-        }
-    }
-
-    async addToSet(hashId: string, key: string, value: string): Promise<void> {
-        const response = await this.safeFetch(`${this.baseUrl}/meta/${hashId}/_add/${key}`, {
-            method: 'POST',
-            headers: { 'Content-Type': 'application/json' },
-            body: JSON.stringify({ value }),
-        });
-        if (response && !response.ok) {
-            console.warn(`[MetaCoreClient] Failed to add to set: ${response.status}`);
-        }
-    }
-
-    async getMetadata(hashId: string): Promise<Record<string, string>> {
+    /**
+     * Whole merged record, in whatever form the endpoint serves (meta-sort's
+     * /meta is nested, meta-core's is flat) — callers flatten. `null` when the
+     * call itself failed, `{}` when the record does not exist.
+     */
+    async getMetadata(hashId: string): Promise<Record<string, unknown> | null> {
         const response = await this.safeFetch(`${this.baseUrl}/meta/${hashId}`, { method: 'GET' });
-        if (!response || response.status === 404) return {};
-        if (!response.ok) return {};
-        const data = await response.json() as { metadata?: Record<string, string> };
+        if (!response) return null;
+        if (response.status === 404) return {};
+        if (!response.ok) return null;
+        const data = await response.json() as { metadata?: Record<string, unknown> };
         return data.metadata ?? {};
     }
 
     /**
-     * Compute the CID (Content Identifier) for a file
-     * Uses the meta-core /file/cid API endpoint
+     * Compute the CID of a file under FILES_PATH via meta-core's `POST /file/cid`.
+     * @param filePath - path relative to FILES_PATH (e.g. "poster/Movie (2024)_poster.jpg")
      */
     async computeFileCID(filePath: string): Promise<string | null> {
-        try {
         const response = await this.safeFetch(`${this.baseUrl}/file/cid`, {
             method: 'POST',
             headers: { 'Content-Type': 'application/json' },
             body: JSON.stringify({ path: filePath }),
         });
         if (!response || !response.ok) return null;
-            const data = await response.json() as { cid?: string };
-            return data.cid ?? null;
-        } catch (error) {
-            console.debug(`[MetaCoreClient] Failed to compute CID for ${filePath}:`, error);
-            return null;
-        }
+        const data = await response.json().catch(() => null) as { cid?: string } | null;
+        return data?.cid ?? null;
     }
 }

src/plugin.ts

@@ -77,6 +77,7 @@ export const manifest: PluginManifest = {
         maxWidth: { type: 'number', label: 'Max Width (px)', default: 640 },
         jpegQuality: { type: 'number', label: 'JPEG Quality (ffmpeg -q:v, lower is better)', default: 4 },
         frameTimeoutMs: { type: 'number', label: 'Per-Frame ffmpeg Timeout (ms)', default: 120000 },
+        maxConcurrent: { type: 'number', label: 'Max Concurrent Extractions', default: 1 },
     },
 };
 
@@ -86,6 +87,20 @@ let seekPercents = [20, 45, 70];
 let maxWidth = 640;
 let jpegQuality = 4;
 let frameTimeoutMs = 120000;
+let maxConcurrent = 1;
+
+/**
+ * A config value as a number, or undefined when it is absent.
+ *
+ * ⚠ Not `Number.isFinite(Number(v))`: `Number(null)` and `Number('')` are both
+ * `0`, so a config payload carrying `"maxWidth": null` would silently set a
+ * 0-pixel width or a 0 ms timeout.
+ */
+function configNumber(v: unknown): number | undefined {
+    if (v === null || v === undefined || v === '') return undefined;
+    const n = Number(v);
+    return Number.isFinite(n) ? n : undefined;
+}
 
 export function configure(config: Record<string, unknown>): void {
     forceRecompute = config.forceRecompute === true;
@@ -96,15 +111,72 @@ export function configure(config: Record<string, unknown>): void {
             .filter((p) => Number.isFinite(p) && p > 0 && p < 100);
         if (parsed.length > 0) seekPercents = parsed;
     }
-    if (Number.isFinite(Number(config.maxWidth))) maxWidth = Number(config.maxWidth);
-    if (Number.isFinite(Number(config.jpegQuality))) jpegQuality = Number(config.jpegQuality);
-    if (Number.isFinite(Number(config.frameTimeoutMs))) frameTimeoutMs = Number(config.frameTimeoutMs);
+    const width = configNumber(config.maxWidth);
+    if (width !== undefined && width > 0) maxWidth = width;
+    const quality = configNumber(config.jpegQuality);
+    if (quality !== undefined && quality >= 1 && quality <= 31) jpegQuality = quality;
+    const timeout = configNumber(config.frameTimeoutMs);
+    if (timeout !== undefined && timeout > 0) frameTimeoutMs = timeout;
+    const concurrent = configNumber(config.maxConcurrent);
+    if (concurrent !== undefined && concurrent >= 1) maxConcurrent = Math.floor(concurrent);
     console.log(
         `[still-extractor] Config: forceRecompute=${forceRecompute}, seekPercents=${seekPercents.join(',')}, ` +
-        `maxWidth=${maxWidth}, jpegQuality=${jpegQuality}, frameTimeoutMs=${frameTimeoutMs}`
+        `maxWidth=${maxWidth}, jpegQuality=${jpegQuality}, frameTimeoutMs=${frameTimeoutMs}, maxConcurrent=${maxConcurrent}`
     );
 }
 
+let activeExtractions = 0;
+const extractionWaiters: Array<() => void> = [];
+
+/**
+ * Run `fn` once an extraction slot is free.
+ *
+ * meta-sort's background queue dispatches several tasks at this single
+ * instance at once, and each ffmpeg grab of a 1080p file peaks near 200 MiB
+ * even with the scale-first chain. Four at a time OOM-killed the 512 MiB
+ * container on the dev stack (`OOMKilled=true`, ffmpeg dying on SIGKILL). The
+ * container gets one CPU, so running them concurrently buys no throughput
+ * anyway — the cap lives here so it holds whatever the scheduler sends.
+ */
+export async function withExtractionSlot<T>(fn: () => Promise<T>): Promise<T> {
+    while (activeExtractions >= maxConcurrent) {
+        await new Promise<void>((resolve) => extractionWaiters.push(resolve));
+    }
+    activeExtractions++;
+    try {
+        return await fn();
+    } finally {
+        activeExtractions--;
+        extractionWaiters.shift()?.();
+    }
+}
+
+/**
+ * Files that produced no usable frame, remembered briefly by cid.
+ *
+ * A failed extraction writes no `still`, so the slot-time re-check cannot stop
+ * the duplicates a replay or rescan queues behind it — on the dev stack a 17s
+ * title-card test clip was fully decoded a dozen times in a row. An in-memory
+ * TTL absorbs that burst without inventing a persisted "tried and failed" key,
+ * and a rescan after the TTL still gets a fresh attempt.
+ */
+const NO_USABLE_FRAME_TTL_MS = 60 * 60 * 1000;
+const noUsableFrameAt = new Map<string, number>();
+
+export function recentlyFailed(cid: string, now = Date.now()): boolean {
+    const at = noUsableFrameAt.get(cid);
+    if (at === undefined) return false;
+    if (now - at > NO_USABLE_FRAME_TTL_MS) {
+        noUsableFrameAt.delete(cid);
+        return false;
+    }
+    return true;
+}
+
+export function rememberNoUsableFrame(cid: string, now = Date.now()): void {
+    noUsableFrameAt.set(cid, now);
+}
+
 /**
  * Compute midhash256 CID from a Buffer (matches meta-hash algorithm).
  * Copied verbatim from metamesh-plugin-subtitle-extractor.
@@ -185,6 +257,32 @@ export interface VideoStream {
     height?: number;
     codec?: string;
     frameRate?: string;
+    /** ffprobe's `attached_pic` disposition. Probe path only — the ffmpeg
+     *  plugin's stream table does not carry dispositions. */
+    attachedPic?: boolean;
+}
+
+/**
+ * `fileinfo/duration` out of whichever shape the payload carries.
+ *
+ * ⚠ meta-sort's /process payload is the NESTED document form — `fileinfo:
+ * { duration }`, `stream: [...]`, `cids: [...]` — not meta-core's flat
+ * `fileinfo/duration` keys. On the dev stack every task's existingMeta had
+ * `fileinfo` and `stream`, never `fileinfo/duration`, and reading only the flat
+ * key sent 16 of 16 stills to the 10s/0s fallback. Accept both, plus a
+ * JSON-string `fileinfo`.
+ */
+export function durationFromMeta(existingMeta: Record<string, unknown> | undefined): number | undefined {
+    if (!existingMeta) return undefined;
+    let fileinfo: unknown = existingMeta['fileinfo'];
+    if (typeof fileinfo === 'string') {
+        try { fileinfo = JSON.parse(fileinfo); } catch { fileinfo = undefined; }
+    }
+    const raw = existingMeta['fileinfo/duration']
+        ?? (fileinfo && typeof fileinfo === 'object' ? (fileinfo as Record<string, unknown>)['duration'] : undefined);
+    if (raw === null || raw === undefined || raw === '') return undefined;
+    const n = Number(raw);
+    return Number.isFinite(n) && n > 0 ? n : undefined;
 }
 
 /**
@@ -210,19 +308,21 @@ const STILL_IMAGE_CODECS = new Set(['mjpeg', 'png', 'bmp', 'gif', 'webp', 'tiff'
  * within the better class; if nothing looks like moving video we fall back to
  * the largest of whatever is there rather than giving up.
  *
- * Accepts both shapes meta-core hands back: the namespaced `stream/{n}` keys
- * the ffmpeg plugin writes, and the flattened `stream` array.
+ * Accepts both shapes: the nested `stream` collection meta-sort's /process
+ * payload actually carries (array, JSON string, or index-keyed object), and the
+ * namespaced `stream/{n}` keys meta-core stores.
  */
 export function selectPrimaryVideoStream(existingMeta: Record<string, string> | undefined): VideoStream | null {
     if (!existingMeta) return null;
 
     const raw: unknown[] = [];
 
-    const flattened = existingMeta['stream'] as unknown;
-    if (flattened) {
+    const nested = existingMeta['stream'] as unknown;
+    if (nested) {
         try {
-            const arr = Array.isArray(flattened) ? flattened : JSON.parse(String(flattened));
-            if (Array.isArray(arr)) raw.push(...arr);
+            const parsed = typeof nested === 'string' ? JSON.parse(nested) : nested;
+            if (Array.isArray(parsed)) raw.push(...parsed);
+            else if (parsed && typeof parsed === 'object') raw.push(...Object.values(parsed as Record<string, unknown>));
         } catch {
             // fall through to the namespaced form
         }
@@ -234,9 +334,7 @@ export function selectPrimaryVideoStream(existingMeta: Record<string, string> |
         }
     }
 
-    const moving: VideoStream[] = [];
-    const other: VideoStream[] = [];
-
+    const candidates: VideoStream[] = [];
     for (const entry of raw) {
         let stream: {
             type?: string; index?: number; width?: number; height?: number;
@@ -249,22 +347,31 @@ export function selectPrimaryVideoStream(existingMeta: Record<string, string> |
         }
         if (!stream || stream.type !== 'video' || stream.index == null) continue;
 
-        const candidate: VideoStream = {
+        candidates.push({
             index: Number(stream.index),
             width: stream.width != null ? Number(stream.width) : undefined,
             height: stream.height != null ? Number(stream.height) : undefined,
             codec: stream.codec,
             frameRate: stream.frameRate,
-        };
-
-        const looksMoving =
-            !!candidate.frameRate && !STILL_IMAGE_CODECS.has((candidate.codec ?? '').toLowerCase());
-        (looksMoving ? moving : other).push(candidate);
+        });
+    }
+    return pickPrimaryVideoStream(candidates);
 }
 
-    const area = (s: VideoStream) => (s.width ?? 0) * (s.height ?? 0);
-    const pool = moving.length > 0 ? moving : other;
-    return pool.reduce<VideoStream | null>((best, s) => (!best || area(s) > area(best) ? s : best), null);
+/**
+ * The moving picture among video candidates, wherever they came from.
+ *
+ * Prefers streams that have a frame rate, are not a still-image codec and are
+ * not flagged as an attached picture; the largest area wins within that class,
+ * falling back to the largest of the rest rather than giving up.
+ */
+export function pickPrimaryVideoStream(candidates: VideoStream[]): VideoStream | null {
+    const looksMoving = (v: VideoStream) =>
+        !v.attachedPic && !!v.frameRate && !STILL_IMAGE_CODECS.has((v.codec ?? '').toLowerCase());
+    const moving = candidates.filter(looksMoving);
+    const pool = moving.length > 0 ? moving : candidates;
+    const area = (v: VideoStream) => (v.width ?? 0) * (v.height ?? 0);
+    return pool.reduce<VideoStream | null>((best, v) => (!best || area(v) > area(best) ? v : best), null);
 }
 
 /**
@@ -323,7 +430,7 @@ function run(
     cmd: string,
     args: string[],
     timeoutMs: number
-): Promise<{ code: number | null; stdout: Buffer; stderr: string; timedOut: boolean }> {
+): Promise<{ code: number | null; signal: NodeJS.Signals | null; stdout: Buffer; stderr: string; timedOut: boolean }> {
     return new Promise((resolve) => {
         const child = spawn(cmd, args, { stdio: ['ignore', 'pipe', 'pipe'] });
         const stdout: Buffer[] = [];
@@ -342,15 +449,74 @@ function run(
         });
         child.on('error', (err) => {
             clearTimeout(timer);
-            resolve({ code: null, stdout: Buffer.concat(stdout), stderr: stderr + String(err), timedOut });
+            resolve({ code: null, signal: null, stdout: Buffer.concat(stdout), stderr: stderr + String(err), timedOut });
         });
-        child.on('close', (code) => {
+        child.on('close', (code, signal) => {
             clearTimeout(timer);
-            resolve({ code, stdout: Buffer.concat(stdout), stderr, timedOut });
+            resolve({ code, signal, stdout: Buffer.concat(stdout), stderr, timedOut });
         });
     });
 }
 
+export interface MediaProbe {
+    duration?: number;
+    videoStreams: VideoStream[];
+}
+
+/**
+ * Duration and video streams read straight from the file.
+ *
+ * Both normally arrive from the ffmpeg plugin, but meta-sort marks a dependency
+ * complete on `failed` and `skipped` callbacks as well as `completed`
+ * (ContainerPluginScheduler.markPluginCompleted), so a still task can run
+ * against a record missing either — or both: on the dev stack a midhash record
+ * had `fileType=video` and no `fileinfo` or `stream` at all. Without a duration
+ * frame choice falls back to 10s/0s (idents and opening credits); without
+ * streams there is nothing to map. One header-only ffprobe answers both.
+ */
+export async function probeMedia(input: string): Promise<MediaProbe> {
+    const args = ['-v', 'error'];
+    if (/^https?:\/\//i.test(input)) {
+        args.push('-rw_timeout', '30000000'); // microseconds
+    }
+    args.push('-print_format', 'json', '-show_format', '-show_streams', input);
+
+    const result = await run('ffprobe', args, 30000);
+    if (result.timedOut || result.code !== 0) return { videoStreams: [] };
+
+    let parsed: {
+        format?: { duration?: string };
+        streams?: Array<{
+            codec_type?: string; index?: number; codec_name?: string;
+            width?: number; height?: number; avg_frame_rate?: string;
+            disposition?: { attached_pic?: number };
+        }>;
+    };
+    try {
+        parsed = JSON.parse(result.stdout.toString());
+    } catch {
+        return { videoStreams: [] };
+    }
+
+    const duration = Number(parsed.format?.duration);
+    const videoStreams = (parsed.streams ?? [])
+        .filter((v) => v.codec_type === 'video' && v.index != null)
+        .map((v): VideoStream => ({
+            index: Number(v.index),
+            width: v.width,
+            height: v.height,
+            codec: v.codec_name,
+            frameRate: v.avg_frame_rate && v.avg_frame_rate !== '0/0' ? v.avg_frame_rate : undefined,
+            attachedPic: v.disposition?.attached_pic === 1,
+        }));
+    return { duration: Number.isFinite(duration) && duration > 0 ? duration : undefined, videoStreams };
+}
+
+/** Container duration read from the file — see probeMedia. */
+export async function probeDuration(input: string): Promise<number | undefined> {
+    return (await probeMedia(input)).duration;
+}
+
 /**
  * Grab one frame at `offsetSec` and return the encoded JPEG.
  *
@@ -365,7 +531,9 @@ export async function grabFrame(
     offsetSec: number,
     outputPath: string
 ): Promise<Buffer | null> {
-    const args = ['-y', '-hide_banner', '-loglevel', 'error'];
+    // -threads caps the decoder's thread pool. Left alone, dav1d/hevc size it
+    // to the host's cores, and every thread holds its own frame buffers.
+    const args = ['-y', '-hide_banner', '-loglevel', 'error', '-threads', '2'];
 
     // rw_timeout is a protocol option; only http/tcp accept it, and passing it
     // to the file protocol makes ffmpeg bail before it starts.
@@ -380,8 +548,11 @@ export async function grabFrame(
         '-an', '-sn', '-dn',
         '-frames:v', '1',
         // thumbnail=N picks the most representative of the next N frames rather
-        // than whatever the seek happened to land on.
-        '-vf', `thumbnail=60,scale='min(iw,${maxWidth})':-2`,
+        // than whatever the seek happened to land on — and it BUFFERS all N.
+        // ⚠ Scale first. `thumbnail=60,scale=…` holds 60 full-resolution frames;
+        // on a 1080p AV1 episode that peaked at 409 MiB, and `scale=…,thumbnail=60`
+        // at 196 MiB (and 2s instead of 4s) for a byte-for-byte comparable JPEG.
+        '-vf', `scale='min(iw,${maxWidth})':-2,thumbnail=60`,
         '-q:v', String(jpegQuality),
         '-f', 'image2',
         outputPath
@@ -393,7 +564,12 @@ export async function grabFrame(
         return null;
     }
     if (result.code !== 0 || !existsSync(outputPath)) {
-        console.warn(`[still-extractor] ffmpeg failed at ${offsetSec}s: ${result.stderr.trim() || `exit ${result.code}`}`);
+        // A signal with no stderr is the kernel, not ffmpeg: in practice the
+        // OOM killer. Say so, instead of the old unhelpful "exit null".
+        const why = result.signal
+            ? `killed by ${result.signal}${result.signal === 'SIGKILL' ? ' (likely out of memory)' : ''}`
+            : result.stderr.trim() || `exit ${result.code}`;
+        console.warn(`[still-extractor] ffmpeg failed at ${offsetSec}s: ${why}`);
         return null;
     }
 
@@ -497,12 +673,16 @@ export async function process(
     const { taskId, cid, filePath, existingMeta } = request;
     const metaCore = new MetaCoreClient(request.metaCoreUrl);
 
-    const skip = (reason: string) =>
-        sendCallback({ taskId, status: 'skipped', duration: Date.now() - startTime, reason });
+    const skip = (reason: string, quiet = false) => {
+        // Every non-video file in the library comes through here too — logging
+        // those would bury the skips that matter ("No video stream" on a video).
+        if (!quiet) console.log(`[still-extractor] ${filePath}: skipped — ${reason}`);
+        return sendCallback({ taskId, status: 'skipped', duration: Date.now() - startTime, reason });
+    };
 
     try {
         if (existingMeta?.fileType !== 'video') {
-            await skip('Not a video file');
+            await skip('Not a video file', true);
             return;
         }
 
@@ -514,16 +694,17 @@ export async function process(
             return;
         }
 
-        const stream = selectPrimaryVideoStream(existingMeta);
-        if (!stream) {
-            await skip('No video stream');
-            return;
-        }
+        let stream = selectPrimaryVideoStream(existingMeta);
 
         // WebDAV of the very core we're enriching, so the frame bytes and the
         // CID we write can never point at different cores.
         const webdavClient = await getWebDAVClient(request.metaCoreUrl);
         if (!webdavClient) {
+            if (!stream) {
+                // Nothing in the record and no way to read the file.
+                await skip('No video stream');
+                return;
+            }
             await sendCallback({
                 taskId,
                 status: 'failed',
@@ -533,10 +714,61 @@ export async function process(
             return;
         }
 
-        const duration = Number(existingMeta?.['fileinfo/duration']);
         const input = webdavClient.toWebDAVUrl(filePath);
+        let probe: MediaProbe | undefined;
+        const probeOnce = async () => (probe ??= await probeMedia(input));
 
-        const extracted = await extractStill(input, stream.index, Number.isFinite(duration) ? duration : undefined);
+        let streamSource = '';
+        if (!stream) {
+            stream = pickPrimaryVideoStream((await probeOnce()).videoStreams);
+            streamSource = ' [probed]';
+            if (!stream) {
+                await skip('No video stream');
+                return;
+            }
+        }
+
+        const recordDuration = durationFromMeta(existingMeta as Record<string, unknown> | undefined);
+
+        if (recordDuration === undefined) {
+            // Name what this task actually received, so an upstream cause of a
+            // missing duration stays diagnosable even though we recover from it.
+            const keys = Object.keys(existingMeta ?? {});
+            console.warn(
+                `[still-extractor] ${filePath}: no usable fileinfo/duration ` +
+                `(fileinfo=${JSON.stringify((existingMeta as Record<string, unknown> | undefined)?.['fileinfo'] ?? existingMeta?.['fileinfo/duration'] ?? null).slice(0, 120)}), probing the file; ` +
+                `existingMeta has ${keys.length} keys: ${keys.slice(0, 40).join(', ')}`
+            );
+        }
+        const duration = recordDuration ?? (await probeOnce()).duration;
+        const durationSource = recordDuration !== undefined ? 'record' : duration !== undefined ? 'probed' : 'unknown';
+
+        console.log(
+            `[still-extractor] ${filePath}: stream=0:${stream.index}${streamSource} ` +
+            `${stream.width ?? '?'}x${stream.height ?? '?'} ${stream.codec ?? ''} ` +
+            `duration=${duration ?? 'unknown'} (${durationSource}) ` +
+            `offsets=${candidateOffsets(duration).join(',')}`
+        );
+
+        const extracted = await withExtractionSlot(async () => {
+            // existingMeta was read at dispatch, before this task queued for a slot.
+            // A replay or rescan queues several tasks per file, so re-check here —
+            // otherwise every duplicate behind the first recomputes the same still.
+            if (!forceRecompute && await metaCore.getProperty(cid, 'still')) return 'present' as const;
+            if (!forceRecompute && recentlyFailed(cid)) return 'recently-failed' as const;
+            const result = await extractStill(input, stream.index, duration);
+            if (!result) rememberNoUsableFrame(cid);
+            return result;
+        });
+        if (extracted === 'present') {
+            await skip('Still already present');
+            return;
+        }
+        if (extracted === 'recently-failed') {
+            // The first attempt already logged "No usable frame".
+            await skip('No usable frame (recent attempt, not retried)', true);
+            return;
+        }
         if (!extracted) {
             await skip('No usable frame');
             return;

test/still.test.ts

@@ -23,6 +23,14 @@ import {
     selectPrimaryVideoStream,
     extractStill,
     grabFrame,
+    configure,
+    withExtractionSlot,
+    probeDuration,
+    durationFromMeta,
+    probeMedia,
+    pickPrimaryVideoStream,
+    recentlyFailed,
+    rememberNoUsableFrame,
 } from '../src/plugin.js';
 import type { CallbackPayload } from '../src/types.js';
 
@@ -282,3 +290,175 @@ describe('process - skip logic', () => {
         expect(seen[0].reason).toBe('No video stream');
     });
 });
+
+describe('withExtractionSlot', () => {
+    const measurePeak = async (jobs: number) => {
+        let active = 0;
+        let peak = 0;
+        const job = () => withExtractionSlot(async () => {
+            active++;
+            peak = Math.max(peak, active);
+            await new Promise((r) => setTimeout(r, 20));
+            active--;
+        });
+        await Promise.all(Array.from({ length: jobs }, job));
+        return peak;
+    };
+
+    it('never runs more extractions than maxConcurrent', async () => {
+        // The dev-stack OOM: four concurrent 1080p grabs in a 512 MiB container.
+        configure({ maxConcurrent: 1 });
+        expect(await measurePeak(4)).toBe(1);
+    });
+
+    it('allows parallelism when configured, and releases every slot', async () => {
+        configure({ maxConcurrent: 3 });
+        expect(await measurePeak(6)).toBe(3);
+        configure({ maxConcurrent: 1 });
+        expect(await measurePeak(3)).toBe(1);
+    });
+
+    it('releases the slot when the job throws', async () => {
+        configure({ maxConcurrent: 1 });
+        await expect(withExtractionSlot(async () => { throw new Error('boom'); })).rejects.toThrow('boom');
+        expect(await measurePeak(2)).toBe(1); // would hang forever if the slot leaked
+    });
+});
+
+describe('configure', () => {
+    it.skipIf(!ffmpegAvailable || !fixturesReady)(
+        'treats null numeric values as absent instead of zero',
+        async () => {
+            // Number(null) === 0: before the fix this set a 0 px width and a 0 ms
+            // timeout, killing every grab on the spot.
+            configure({ maxWidth: null, jpegQuality: null, frameTimeoutMs: null, maxConcurrent: null });
+            mkdirSync(TEMP, { recursive: true });
+            const result = await extractStill(fixture('color.mp4'), 0, 10, TEMP);
+            expect(result).not.toBeNull();
+        }
+    );
+});
+
+describe.skipIf(!ffmpegAvailable || !fixturesReady)('probeDuration', () => {
+    it('reads the duration from the file header', async () => {
+        const d = await probeDuration(fixture('color.mp4'));
+        expect(d).toBeGreaterThan(9);
+        expect(d).toBeLessThan(11);
+    });
+
+    it('returns undefined for an unreadable input instead of throwing', async () => {
+        expect(await probeDuration(fixture('does-not-exist.mkv'))).toBeUndefined();
+    });
+
+    it('turns a record with no duration into real candidate offsets, not the 10s/0s fallback', async () => {
+        const d = await probeDuration(fixture('black-head.mp4'));
+        expect(d).toBeDefined();
+        expect(candidateOffsets(d)).not.toEqual([10, 0]);
+        // and the 20% candidate lands past the 6s black head
+        expect(candidateOffsets(d)[0]).toBeGreaterThanOrEqual(4);
+    });
+});
+
+describe('durationFromMeta', () => {
+    it('reads the nested document form meta-sort actually sends', () => {
+        expect(durationFromMeta({ fileType: 'video', fileinfo: { duration: '1433.089', formatName: 'matroska,webm' } }))
+            .toBeCloseTo(1433.089);
+    });
+
+    it('reads a numeric nested duration, the exact shape metasort-app/meta returns', () => {
+        // Captured from the dev stack: fileinfo.duration is a JSON number there, not a string.
+        expect(durationFromMeta({ fileinfo: { duration: 1433.089, formatName: 'matroska,webm' } })).toBe(1433.089);
+    });
+
+    it('reads the flat meta-core key', () => {
+        expect(durationFromMeta({ 'fileinfo/duration': '1380.031' })).toBeCloseTo(1380.031);
+    });
+
+    it('reads a JSON-string fileinfo', () => {
+        expect(durationFromMeta({ fileinfo: JSON.stringify({ duration: '42.5' }) })).toBe(42.5);
+    });
+
+    it('treats missing, zero and junk values as absent', () => {
+        expect(durationFromMeta(undefined)).toBeUndefined();
+        expect(durationFromMeta({})).toBeUndefined();
+        expect(durationFromMeta({ fileinfo: {} })).toBeUndefined();
+        expect(durationFromMeta({ fileinfo: { duration: '0' } })).toBeUndefined();
+        expect(durationFromMeta({ fileinfo: { duration: 'N/A' } })).toBeUndefined();
+        expect(durationFromMeta({ fileinfo: 'not json' })).toBeUndefined();
+    });
+});
+
+describe('selectPrimaryVideoStream — nested payload shapes', () => {
+    it('reads a nested stream object keyed by index', () => {
+        const picked = selectPrimaryVideoStream({
+            stream: {
+                '0': JSON.stringify({ type: 'audio', index: 0 }),
+                '1': JSON.stringify({ type: 'video', index: 1, width: 1280, height: 720, codec: 'h264', frameRate: '24/1' }),
+            },
+        } as unknown as Record<string, string>);
+        expect(picked?.index).toBe(1);
+    });
+
+    it('reads a nested stream array of objects', () => {
+        const picked = selectPrimaryVideoStream({
+            stream: [
+                { type: 'video', index: 0, width: 1920, height: 1080, codec: 'av1', frameRate: '24000/1001' },
+                { type: 'audio', index: 1 },
+            ],
+        } as unknown as Record<string, string>);
+        expect(picked?.index).toBe(0);
+    });
+});
+
+describe('pickPrimaryVideoStream', () => {
+    it('never prefers an attached picture, even one with a frame rate', () => {
+        const picked = pickPrimaryVideoStream([
+            { index: 0, width: 640, height: 480, codec: 'h264', frameRate: '24/1' },
+            { index: 1, width: 1000, height: 1500, codec: 'h264', frameRate: '24/1', attachedPic: true },
+        ]);
+        expect(picked?.index).toBe(0);
+    });
+
+    it('returns null with no candidates', () => {
+        expect(pickPrimaryVideoStream([])).toBeNull();
+    });
+});
+
+describe.skipIf(!ffmpegAvailable || !fixturesReady)('probeMedia', () => {
+    it('reads the duration and video stream of a plain video', async () => {
+        const probe = await probeMedia(fixture('color.mp4'));
+        expect(probe.duration).toBeGreaterThan(9);
+        expect(pickPrimaryVideoStream(probe.videoStreams)?.width).toBe(320);
+    });
+
+    it('flags attached cover art and never picks it', async () => {
+        const probe = await probeMedia(fixture('cover-art.mp4'));
+        expect(probe.videoStreams.some((v) => v.attachedPic)).toBe(true);
+        const primary = pickPrimaryVideoStream(probe.videoStreams);
+        expect(primary?.attachedPic).toBe(false);
+        expect(primary?.width).toBe(320);
+    });
+
+    it('finds no video stream in audio-only input', async () => {
+        expect(pickPrimaryVideoStream((await probeMedia(fixture('audio-only.mp3'))).videoStreams)).toBeNull();
+    });
+
+    it('returns an empty probe for unreadable input instead of throwing', async () => {
+        expect(await probeMedia(fixture('does-not-exist.mkv'))).toEqual({ videoStreams: [] });
+    });
+});
+
+describe('no-usable-frame memory', () => {
+    it('remembers a failed cid for an hour, then forgets it', () => {
+        const t0 = 1_000_000;
+        expect(recentlyFailed('bafk-fail', t0)).toBe(false);
+        rememberNoUsableFrame('bafk-fail', t0);
+        expect(recentlyFailed('bafk-fail', t0 + 59 * 60 * 1000)).toBe(true);
+        expect(recentlyFailed('bafk-fail', t0 + 61 * 60 * 1000)).toBe(false);
+    });
+
+    it('is keyed per cid', () => {
+        rememberNoUsableFrame('bafk-a', 5);
+        expect(recentlyFailed('bafk-b', 5)).toBe(false);
+    });
+});