Merge pull request 'Agent: short-video sampling fix + "Retry errored jobs" recovery action' (#185) from dev into main

agent/fc_agent/app.py

@@ -21,7 +21,7 @@ log = logging.getLogger("fc_agent.app")
 # Bump on every agent change. The page embeds this and /status reports it; the UI
 # warns to reload when they differ — so a stale browser-cached page can't be
 # mistaken for "the new image didn't deploy". (Belt-and-braces with no-store.)
-VERSION = "2026-07-01.10 · fix Status freeze (conchint.textContent destroyed #capn)"
+VERSION = "2026-07-02.1 · short videos sample again (select, not fps) + ffmpeg errors logged"
 
 logbuf.install()
 cfg = Config.from_env()

agent/fc_agent/media.py

@@ -2,6 +2,7 @@
 (ffmpeg) at the cadence FC sends — so a video becomes a bag of per-frame
 instances, each with a timestamp."""
 import io
+import logging
 import os
 import subprocess
 import tempfile
@@ -9,6 +10,8 @@ import time
 
 from PIL import Image, ImageFile
 
+log = logging.getLogger("fc_agent.media")
+
 # Load slightly-truncated images (a few missing trailing bytes) instead of
 # raising — matches the server embedder. These are common in scraped libraries
 # and would otherwise fail the job 3× then error (operator-flagged 2026-06-30).
@@ -134,29 +137,62 @@ def sample_frames_from_url(
     interval = max(0.5, float(interval_seconds or 4.0))
     cap = max(1, int(max_frames or 64))
     hdr = ["-headers", headers] if headers else []
+    # select (NOT the fps filter): always keep the FIRST frame, then one per
+    # `interval` seconds of timestamp. fps=1/N emits round(duration/N) frames,
+    # which is ZERO for any clip shorter than ~N/2 seconds — a whole class of
+    # short animation loops failed as "unprocessable" that way (operator-flagged
+    # 2026-07-02: 0.5s/1.75s clips). scale=out_range=full converts limited-range
+    # yuv420p to full range so the mjpeg (jpg) encoder accepts it at default
+    # strictness instead of erroring on "non full-range YUV".
+    vf = (
+        f"select='isnan(prev_selected_t)+gte(t-prev_selected_t\\,{interval})',"
+        "scale=out_range=full"
+    )
     with tempfile.TemporaryDirectory() as tmp:
         pattern = os.path.join(tmp, "f_%05d.jpg")
         cmd = ["ffmpeg", "-nostdin", "-loglevel", "error", *_RECONNECT, *hdr,
-               "-i", url, "-vf", f"fps=1/{interval}", "-frames:v", str(cap),
-               "-q:v", "3", pattern]
+               "-i", url, "-vf", vf, "-fps_mode", "vfr",
+               "-frames:v", str(cap), "-q:v", "3", pattern]
+        # ffmpeg's stderr goes to a file (not a PIPE, which could fill and
+        # deadlock; not DEVNULL, which is how a filter bug hid as "unprocessable"
+        # for weeks) — on failure its tail is logged so the operator can see WHY.
+        errpath = os.path.join(tmp, "stderr.txt")
         try:
-            proc = subprocess.Popen(
-                cmd, stdin=subprocess.DEVNULL,
-                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
-            )
+            with open(errpath, "wb") as errf:
+                proc = subprocess.Popen(
+                    cmd, stdin=subprocess.DEVNULL,
+                    stdout=subprocess.DEVNULL, stderr=errf,
+                )
+                # Poll rather than block, so a Stop (or the per-video timeout) can
+                # kill a slow/wedged ffmpeg promptly instead of waiting it out.
+                start = time.monotonic()
+                while True:
+                    try:
+                        proc.wait(timeout=0.5)
+                        break
+                    except subprocess.TimeoutExpired:
+                        stopped = should_stop and should_stop()
+                        if stopped or (time.monotonic() - start > timeout):
+                            _terminate(proc)
+                            if not stopped:
+                                log.warning("ffmpeg timed out after %.0fs: %s",
+                                            timeout, url)
+                            return []
         except (OSError, ValueError):
             return []
-        # Poll rather than block, so a Stop (or the per-video timeout) can kill a
-        # slow/wedged ffmpeg promptly instead of waiting it out.
-        start = time.monotonic()
-        while True:
-            try:
-                proc.wait(timeout=0.5)
-                break
-            except subprocess.TimeoutExpired:
-                if (should_stop and should_stop()) or (time.monotonic() - start > timeout):
-                    _terminate(proc)
-                    return []
-        if proc.returncode != 0:
-            return []
-        return _collect_frames(tmp, interval, cap)
+        frames = _collect_frames(tmp, interval, cap)
+        if not frames:
+            log.warning("ffmpeg produced no frames (exit %s) for %s — stderr: %s",
+                        proc.returncode, url, _tail(errpath))
+        return frames
+
+
+def _tail(path: str, limit: int = 300) -> str:
+    """Last `limit` chars of a (stderr) file, flattened — for failure logs."""
+    try:
+        with open(path, "rb") as f:
+            f.seek(0, os.SEEK_END)
+            f.seek(max(0, f.tell() - limit))
+            return f.read().decode("utf-8", "replace").replace("\n", " ").strip()
+    except OSError:
+        return "?"

backend/app/api/gpu.py

@@ -11,7 +11,7 @@ homelab admin.
 import secrets
 
 from quart import Blueprint, jsonify, request
-from sqlalchemy import func, select
+from sqlalchemy import func, select, update
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 
 from ..extensions import get_session
@@ -109,6 +109,27 @@ async def reprocess():
     return jsonify({"celery_task_id": r.id, "task": task}), 202
 
 
+@gpu_bp.route("/retry_errors", methods=["POST"])
+async def retry_errors():
+    """Requeue every ERRORED job (all task types) back to pending — the scoped
+    recovery after an agent-side fix (e.g. the short-video sampler), where
+    /reprocess would needlessly re-run the whole done library too. Attempts and
+    the stored error reset so each job gets its full retry budget under the
+    fixed pipeline. Small row count (errors only) → inline UPDATE, and the
+    response carries the number requeued for the UI toast."""
+    async with get_session() as session:
+        res = await session.execute(
+            update(GpuJob)
+            .where(GpuJob.status == "error")
+            .values(
+                status="pending", attempts=0, error=None, lease_token=None,
+                leased_at=None, lease_expires_at=None, updated_at=func.now(),
+            )
+        )
+        await session.commit()
+    return jsonify({"requeued": res.rowcount or 0})
+
+
 # --- Agent (bearer token): lease / submit / heartbeat / fail ------------
 
 @gpu_bp.route("/jobs/lease", methods=["POST"])

frontend/src/components/settings/GpuAgentCard.vue

@@ -52,6 +52,17 @@
       <div class="fc-q"><div class="fc-q__n" :class="queue.error ? 'fc-weak' : ''">{{ queue.error }}</div><div class="fc-q__l">errored</div></div>
     </div>
 
+    <v-btn
+      class="mt-3" color="accent" variant="tonal" rounded="pill" size="small"
+      prepend-icon="mdi-restart-alert" :loading="retrying"
+      :disabled="!queue.error" @click="onRetryErrors"
+    >Retry errored jobs</v-btn>
+    <p class="fc-muted text-caption mt-2 mb-0">
+      Errored jobs park after 3 failed attempts. This requeues just those (their
+      errors cleared, attempts reset) — use after updating the agent, without
+      re-running the whole done library.
+    </p>
+
     <v-btn
       class="mt-4" color="accent" variant="tonal" rounded="pill" size="small"
       prepend-icon="mdi-account-box-multiple" :loading="backfilling" @click="onBackfill"
@@ -164,6 +175,7 @@ const rotating = ref(false)
 const backfilling = ref(false)
 const backfillingSiglip = ref(false)
 const reprocessing = ref(false)
+const retrying = ref(false)
 const threshold = ref(0.85)
 const savingThreshold = ref(false)
 const autoApply = ref(true)
@@ -323,6 +335,19 @@ async function onBackfillSiglip() {
   }
 }
 
+async function onRetryErrors() {
+  retrying.value = true
+  try {
+    const { requeued } = await store.retryErrors()
+    toast({ text: `Requeued ${requeued} errored job${requeued === 1 ? '' : 's'} — run the agent to process them`, type: 'success' })
+    await refreshQueue()
+  } catch (e) {
+    toast({ text: `Could not retry errored jobs: ${e.message}`, type: 'error' })
+  } finally {
+    retrying.value = false
+  }
+}
+
 async function onReprocess() {
   if (!window.confirm('Re-process the ENTIRE library (re-detect + re-crop every image)? This is heavy and runs on the GPU agent.')) return
   reprocessing.value = true

frontend/src/stores/gpu.js

@@ -35,5 +35,11 @@ export const useGpuStore = defineStore('gpu', () => {
     return await api.post('/api/gpu/reprocess', { body: { task } })
   }
 
-  return { token, rotateToken, status, backfill, reprocess }
+  // Requeue ONLY the errored jobs (all task types) — the scoped recovery after
+  // an agent fix, without re-running the done library. Returns { requeued }.
+  async function retryErrors() {
+    return await api.post('/api/gpu/retry_errors')
+  }
+
+  return { token, rotateToken, status, backfill, reprocess, retryErrors }
 })

tests/test_api_gpu.py

@@ -1,7 +1,8 @@
 """GPU-job HTTP API (#114): bearer auth + lease/submit round-trip + backfill."""
 import pytest
+from sqlalchemy import select
 
-from backend.app.models import ImageRecord
+from backend.app.models import GpuJob, ImageRecord
 from backend.app.services.ml.gpu_jobs import GpuJobService
 from backend.app.services.ml.regions import RegionService
 
@@ -148,3 +149,39 @@ async def test_release_hands_job_back_to_pending(client, db):
     assert resp.status_code == 200 and (await resp.get_json())["released"] == 1
     st = await (await client.get("/api/gpu/status")).get_json()
     assert st["pending"] == 1 and st["leased"] == 0
+
+
+@pytest.mark.asyncio
+async def test_retry_errors_requeues_only_errored(client, db):
+    """/retry_errors resets ERRORED jobs (any task) to pending with a fresh
+    retry budget — and leaves done work untouched (it is NOT /reprocess)."""
+    img1 = await _img(db, "1" * 64)
+    img2 = await _img(db, "2" * 64)
+    svc = GpuJobService(db)
+    j_err = await svc.enqueue(img1.id, "ccip")
+    j_done = await svc.enqueue(img2.id, "siglip")
+    err_id = j_err.id
+    done_id = j_done.id
+    j_err.status = "error"
+    j_err.attempts = 3
+    j_err.error = "no frames sampled from video (unprocessable)"
+    j_done.status = "done"
+    await db.commit()
+
+    resp = await client.post("/api/gpu/retry_errors")
+    assert resp.status_code == 200
+    assert (await resp.get_json())["requeued"] == 1
+
+    # Column selects, not ORM refresh — the route wrote via Core DML.
+    row = (await db.execute(
+        select(GpuJob.status, GpuJob.attempts, GpuJob.error)
+        .where(GpuJob.id == err_id)
+    )).one()
+    assert tuple(row) == ("pending", 0, None)
+    done_status = await db.scalar(
+        select(GpuJob.status).where(GpuJob.id == done_id)
+    )
+    assert done_status == "done"
+
+    st = await (await client.get("/api/gpu/status")).get_json()
+    assert st["pending"] == 1 and st["error"] == 0