From b1b129ce9f08e7d95038d9aa6e6765f7809a2d45 Mon Sep 17 00:00:00 2001
From: Bryan Van Deusen <bryan.vandeusen@fabledsword.com>
Date: Wed, 27 May 2026 22:18:02 -0400
Subject: [PATCH 1/7] =?UTF-8?q?feat(downloads-tab):=20A+B=20dashboard=20im?=
 =?UTF-8?q?provements=20=E2=80=94=20row=20restyle=20+=20date-grouped=20sec?=
 =?UTF-8?q?tions=20with=20failed-pinned?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Operator-flagged 2026-05-27: the Downloads subtab "doesn't feel like a
dashboard" — status was a tiny mdi icon at the far left, platform chip
was neutral-tonal, errors were plain orange text floating on the right,
and all 28 rows from the same hour visually had the same priority.

**Row restyle (A):**
- 4px colored left-edge bar by status (success/error/info/warning/grey)
  — visually scannable at the edge without parsing the chip text
- Status chip with text label (Completed/Failed/Running/Queued/Skipped)
  + leading icon, tonal-colored. Replaces the bare mdi-icon.
- Platform chip swapped to the color-coded subscriptions/PlatformChip
  (Patreon=red mdi-patreon, SubscribeStar=amber, HentaiFoundry=purple,
  Discord=indigo, Pixiv=blue, DeviantArt=green).
- File count: tonal info chip when > 0, dim middle-dot when 0 (so
  scheduled "no-change" scans don't dominate the column visually).
- Error: red tonal pill chip with leading icon, truncated to 60 chars
  with full text in the title tooltip. Replaces plain text.
- Per-row actions (hidden at 50% opacity, fade to full on row hover):
  Retry (only when status=error AND source_id known — hits
  POST /api/sources/<id>/check via the existing sources.checkNow),
  Details (opens the detail modal), Open artist (navigates to the
  artist page). Clicks stop-propagation so they don't bubble to the
  row click.

**Date-grouped sections (B):**
- Events are bucketed into four sections: Today / Yesterday /
  Last 7 days / Earlier. Empty buckets are skipped. Buckets boundaries
  are computed against the operator's local-time start-of-day so
  "Today" matches their intuition.
- Each section has a collapsible header with a row-count chip + a
  red "failed in this section" chip when any failures are in scope.
- Within each section, status='error' rows are pinned to the top
  (operator's eye lands on failures first; successful scans flow
  below).
- Collapsed state persists across refresh within the SubscriptionsView
  lifetime (reactive object, default all-expanded).

DownloadEventRow grid widened to accommodate the status chip + actions
column. PolyMasonry-style ellipsis on the artist link prevents long
names from breaking the layout.

No new endpoints; the Retry path reuses the existing /api/sources/<id>/check
flow (the source-check endpoint was already in place, just not wired
into a per-row button).
---
 .../components/downloads/DownloadEventRow.vue | 216 +++++++++++++++---
 .../components/subscriptions/DownloadsTab.vue | 118 +++++++++-
 2 files changed, 292 insertions(+), 42 deletions(-)

diff --git a/frontend/src/components/downloads/DownloadEventRow.vue b/frontend/src/components/downloads/DownloadEventRow.vue
index a4360c0..fb2999b 100644
--- a/frontend/src/components/downloads/DownloadEventRow.vue
+++ b/frontend/src/components/downloads/DownloadEventRow.vue
@@ -1,47 +1,118 @@
 <template>
-  <div class="fc-dl-row" @click="$emit('open', event.id)">
-    <v-icon :icon="statusIcon" :color="statusColor" size="small" />
+  <div
+    class="fc-dl-row"
+    :class="[`fc-dl-row--${event.status || 'unknown'}`]"
+    @click="$emit('open', event.id)"
+  >
+    <!-- Colored left edge marks the run's status; matches the row's
+         status-chip color but reads at a glance without needing to
+         parse the chip text. -->
+    <div class="fc-dl-row__bar" />
+
+    <v-chip
+      :color="statusColor"
+      size="small"
+      variant="tonal"
+      :prepend-icon="statusIcon"
+      class="fc-dl-row__status"
+    >{{ statusLabel }}</v-chip>
+
     <RouterLink
       v-if="event.artist_slug"
       :to="`/artist/${event.artist_slug}`"
       class="fc-dl-row__artist"
       @click.stop
     >{{ event.artist_name }}</RouterLink>
-    <span v-else class="fc-dl-row__artist">—</span>
-    <v-chip size="x-small" variant="tonal">{{ event.platform || '—' }}</v-chip>
-    <span class="fc-dl-row__time">{{ fmtTime(event.started_at) }}</span>
-    <span class="fc-dl-row__files">{{ event.files_count }} files</span>
-    <span class="fc-dl-row__duration">{{ fmtDuration(event.summary?.duration_seconds) }}</span>
-    <span v-if="event.error" class="fc-dl-row__error">{{ event.error }}</span>
+    <span v-else class="fc-dl-row__artist fc-dl-row__artist--missing">—</span>
+
+    <PlatformChip
+      v-if="event.platform"
+      :platform="event.platform"
+      size="x-small"
+      class="fc-dl-row__platform"
+    />
+    <span v-else class="fc-dl-row__platform-missing">—</span>
+
+    <span class="fc-dl-row__time" :title="event.started_at">
+      {{ fmtTime(event.started_at) }}
+    </span>
+
+    <v-chip
+      v-if="event.files_count > 0"
+      size="x-small" variant="tonal" color="info"
+      prepend-icon="mdi-image-multiple"
+      class="fc-dl-row__files"
+    >{{ event.files_count }}</v-chip>
+    <span v-else class="fc-dl-row__no-files" aria-label="no new files">·</span>
+
+    <span class="fc-dl-row__duration">
+      {{ fmtDuration(event.summary?.duration_seconds) }}
+    </span>
+
+    <v-chip
+      v-if="event.error"
+      color="error" size="x-small" variant="tonal"
+      prepend-icon="mdi-alert-octagon"
+      class="fc-dl-row__error"
+      :title="event.error"
+    >{{ truncateError(event.error) }}</v-chip>
+    <span v-else class="fc-dl-row__error-spacer" />
+
+    <div class="fc-dl-row__actions" @click.stop>
+      <v-btn
+        v-if="event.status === 'error' && event.source_id"
+        icon size="x-small" variant="text" color="warning"
+        :loading="retrying"
+        @click.stop="onRetry"
+      >
+        <v-icon size="small">mdi-refresh</v-icon>
+        <v-tooltip activator="parent" location="top">Retry source check</v-tooltip>
+      </v-btn>
+      <v-btn icon size="x-small" variant="text" @click.stop="$emit('open', event.id)">
+        <v-icon size="small">mdi-information-outline</v-icon>
+        <v-tooltip activator="parent" location="top">Details</v-tooltip>
+      </v-btn>
+      <v-btn
+        v-if="event.artist_slug"
+        icon size="x-small" variant="text"
+        :to="`/artist/${event.artist_slug}`"
+        @click.stop
+      >
+        <v-icon size="small">mdi-account-circle</v-icon>
+        <v-tooltip activator="parent" location="top">Open artist</v-tooltip>
+      </v-btn>
+    </div>
   </div>
 </template>
 
 <script setup>
-import { computed } from 'vue'
+import { computed, ref } from 'vue'
 import { RouterLink } from 'vue-router'
 
+import PlatformChip from '../subscriptions/PlatformChip.vue'
+import { useSourcesStore } from '../../stores/sources.js'
+
 const props = defineProps({ event: { type: Object, required: true } })
 defineEmits(['open'])
 
-const statusIcon = computed(() => ({
-  ok: 'mdi-check-circle',
-  error: 'mdi-alert-circle',
-  running: 'mdi-progress-clock',
-  pending: 'mdi-clock-outline',
-  skipped: 'mdi-minus-circle',
-}[props.event.status] || 'mdi-help-circle'))
+const sourcesStore = useSourcesStore()
+const retrying = ref(false)
 
-const statusColor = computed(() => ({
-  ok: 'success',
-  error: 'error',
-  running: 'info',
-  pending: 'secondary',
-  skipped: 'warning',
-}[props.event.status] || undefined))
+const _STATUS = {
+  ok:      { color: 'success', icon: 'mdi-check-circle',    label: 'Completed' },
+  error:   { color: 'error',   icon: 'mdi-alert-circle',    label: 'Failed' },
+  running: { color: 'info',    icon: 'mdi-progress-clock',  label: 'Running' },
+  pending: { color: 'grey',    icon: 'mdi-clock-outline',   label: 'Queued' },
+  skipped: { color: 'warning', icon: 'mdi-skip-next',       label: 'Skipped' },
+}
+const statusColor = computed(() => _STATUS[props.event.status]?.color || 'grey')
+const statusIcon = computed(() => _STATUS[props.event.status]?.icon || 'mdi-help-circle')
+const statusLabel = computed(() => _STATUS[props.event.status]?.label || props.event.status)
 
 function fmtTime(iso) {
   if (!iso) return '—'
-  return iso.slice(0, 19).replace('T', ' ')
+  // 2026-05-27 23:36 — second granularity is in the row's title attr
+  return iso.slice(0, 16).replace('T', ' ')
 }
 function fmtDuration(sec) {
   if (sec == null) return '—'
@@ -49,34 +120,107 @@ function fmtDuration(sec) {
   const m = Math.floor(sec / 60), s = Math.floor(sec % 60)
   return `${m}m ${s}s`
 }
+function truncateError(msg) {
+  const s = String(msg || '')
+  if (s.length <= 60) return s
+  return s.slice(0, 57) + '…'
+}
+
+async function onRetry() {
+  if (!props.event.source_id) return
+  retrying.value = true
+  try {
+    await sourcesStore.checkNow(props.event.source_id)
+    globalThis.window?.__fcToast?.({
+      text: `Source check re-queued`, type: 'success',
+    })
+  } catch (e) {
+    const isInFlight = !!e?.body?.download_event_id
+    globalThis.window?.__fcToast?.({
+      text: isInFlight ? 'Already running' : `Retry failed: ${e?.detail || e?.message || e}`,
+      type: isInFlight ? 'info' : 'error',
+    })
+  } finally {
+    retrying.value = false
+  }
+}
 </script>
 
 <style scoped>
 .fc-dl-row {
+  position: relative;
   display: grid;
-  grid-template-columns: 24px 1fr 96px 160px 80px 80px 1fr;
-  gap: 0.75rem;
+  grid-template-columns:
+    /* bar */    4px
+    /* status */ 120px
+    /* artist */ minmax(120px, 1.2fr)
+    /* plat   */ 140px
+    /* time   */ 140px
+    /* files  */ 60px
+    /* dur    */ 70px
+    /* error  */ minmax(0, 1.5fr)
+    /* actions*/ 120px;
+  gap: 0.6rem;
   align-items: center;
-  padding: 0.5rem 0.75rem;
-  border-bottom: 1px solid rgb(var(--v-theme-on-surface-variant) / 0.18);
+  padding: 0.55rem 0.75rem 0.55rem 0;
+  border-bottom: 1px solid rgb(var(--v-theme-on-surface-variant) / 0.15);
   cursor: pointer;
+  transition: background 0.12s ease;
 }
-.fc-dl-row:hover { background: rgb(var(--v-theme-surface) / 0.5); }
+.fc-dl-row:hover {
+  background: rgb(var(--v-theme-on-surface) / 0.04);
+}
+.fc-dl-row__bar {
+  width: 4px;
+  align-self: stretch;
+  border-radius: 0 2px 2px 0;
+}
+.fc-dl-row--ok      .fc-dl-row__bar { background: rgb(var(--v-theme-success)); }
+.fc-dl-row--error   .fc-dl-row__bar { background: rgb(var(--v-theme-error));   }
+.fc-dl-row--running .fc-dl-row__bar { background: rgb(var(--v-theme-info));    }
+.fc-dl-row--skipped .fc-dl-row__bar { background: rgb(var(--v-theme-warning)); }
+.fc-dl-row--pending .fc-dl-row__bar { background: rgb(var(--v-theme-on-surface-variant) / 0.4); }
+
+.fc-dl-row__status { justify-self: start; }
 .fc-dl-row__artist {
   color: rgb(var(--v-theme-on-surface));
   text-decoration: none;
+  font-weight: 500;
+  overflow: hidden; text-overflow: ellipsis; white-space: nowrap;
+}
+.fc-dl-row__artist--missing,
+.fc-dl-row__platform-missing,
+.fc-dl-row__no-files {
+  color: rgb(var(--v-theme-on-surface-variant));
+  opacity: 0.5;
 }
 .fc-dl-row__artist:hover { color: rgb(var(--v-theme-accent)); }
-.fc-dl-row__time, .fc-dl-row__files, .fc-dl-row__duration {
+.fc-dl-row__platform { justify-self: start; }
+.fc-dl-row__time,
+.fc-dl-row__duration {
   color: rgb(var(--v-theme-on-surface-variant));
   font-size: 0.85rem;
   font-variant-numeric: tabular-nums;
-}
-.fc-dl-row__error {
-  color: rgb(var(--v-theme-error));
-  font-size: 0.85rem;
-  overflow: hidden;
-  text-overflow: ellipsis;
   white-space: nowrap;
 }
+.fc-dl-row__no-files {
+  text-align: center;
+  font-size: 1.1rem;
+}
+.fc-dl-row__error {
+  justify-self: start;
+  max-width: 100%;
+}
+.fc-dl-row__error :deep(.v-chip__content) {
+  overflow: hidden; text-overflow: ellipsis; white-space: nowrap;
+}
+.fc-dl-row__error-spacer { /* keeps the grid column reserved */ }
+
+.fc-dl-row__actions {
+  display: flex; gap: 2px;
+  justify-self: end;
+  opacity: 0.5;
+  transition: opacity 0.12s ease;
+}
+.fc-dl-row:hover .fc-dl-row__actions { opacity: 1; }
 </style>
diff --git a/frontend/src/components/subscriptions/DownloadsTab.vue b/frontend/src/components/subscriptions/DownloadsTab.vue
index 7ced798..ecaa92b 100644
--- a/frontend/src/components/subscriptions/DownloadsTab.vue
+++ b/frontend/src/components/subscriptions/DownloadsTab.vue
@@ -20,15 +20,44 @@
       <v-progress-circular indeterminate color="accent" size="36" />
     </div>
 
-    <div v-else-if="store.events.length === 0" class="fc-dl__empty">
+    <div v-else-if="filteredEvents.length === 0" class="fc-dl__empty">
       <p>No download events match the current filter.</p>
     </div>
 
     <div v-else>
-      <DownloadEventRow
-        v-for="e in filteredEvents" :key="e.id" :event="e"
-        @open="openDetail"
-      />
+      <section
+        v-for="g in groups" :key="g.key"
+        class="fc-dl__group"
+      >
+        <header
+          class="fc-dl__group-head"
+          role="button" tabindex="0"
+          @click="toggle(g.key)" @keydown.enter="toggle(g.key)"
+        >
+          <v-icon size="small" class="fc-dl__group-chev">
+            {{ collapsed[g.key] ? 'mdi-chevron-right' : 'mdi-chevron-down' }}
+          </v-icon>
+          <span class="fc-dl__group-label">{{ g.label }}</span>
+          <span class="fc-dl__group-counts">
+            <v-chip
+              v-if="g.failedCount > 0"
+              size="x-small" color="error" variant="tonal"
+              prepend-icon="mdi-alert-circle"
+            >{{ g.failedCount }}</v-chip>
+            <v-chip size="x-small" variant="tonal">
+              {{ g.items.length }}
+              {{ g.items.length === 1 ? 'event' : 'events' }}
+            </v-chip>
+          </span>
+        </header>
+        <div v-if="!collapsed[g.key]" class="fc-dl__group-body">
+          <DownloadEventRow
+            v-for="e in g.items" :key="e.id" :event="e"
+            @open="openDetail"
+          />
+        </div>
+      </section>
+
       <div class="fc-dl__sentinel">
         <v-btn v-if="store.hasMore" variant="text" @click="store.loadMore()" :loading="store.loading">
           Load more
@@ -45,7 +74,7 @@
 </template>
 
 <script setup>
-import { computed, onMounted, ref, watch } from 'vue'
+import { computed, onMounted, reactive, ref, watch } from 'vue'
 import { useRoute } from 'vue-router'
 
 import { useDownloadsStore } from '../../stores/downloads.js'
@@ -59,6 +88,16 @@ const route = useRoute()
 const store = useDownloadsStore()
 const filterModel = ref({ ...store.filter })
 
+// Each group's collapsed state persists across refreshes for the
+// lifetime of the SubscriptionsView (operator-friendly default: all
+// expanded; collapse what you don't care about right now).
+const collapsed = reactive({
+  today: false, yesterday: false, week: false, earlier: false,
+})
+function toggle(key) {
+  collapsed[key] = !collapsed[key]
+}
+
 async function refresh() {
   await Promise.all([
     store.loadFirst(),
@@ -91,6 +130,48 @@ const filteredEvents = computed(() => {
   return arr
 })
 
+// Group events by relative date bucket and pin failed runs to the
+// top of each bucket. Buckets boundaries are computed against the
+// operator's local-time start-of-day so "Today" matches their
+// intuition regardless of the event's stored UTC timestamp.
+const groups = computed(() => {
+  const now = new Date()
+  const startOfToday = new Date(
+    now.getFullYear(), now.getMonth(), now.getDate(),
+  ).getTime()
+  const startOfYesterday = startOfToday - 24 * 3600 * 1000
+  const startOfWeek = startOfToday - 7 * 24 * 3600 * 1000
+
+  const buckets = { today: [], yesterday: [], week: [], earlier: [] }
+  for (const e of filteredEvents.value) {
+    const t = new Date(e.started_at).getTime()
+    if (t >= startOfToday) buckets.today.push(e)
+    else if (t >= startOfYesterday) buckets.yesterday.push(e)
+    else if (t >= startOfWeek) buckets.week.push(e)
+    else buckets.earlier.push(e)
+  }
+
+  function withFailedPinned(items) {
+    const fail = items.filter((e) => e.status === 'error')
+    const rest = items.filter((e) => e.status !== 'error')
+    return [...fail, ...rest]
+  }
+
+  const meta = [
+    { key: 'today', label: 'Today' },
+    { key: 'yesterday', label: 'Yesterday' },
+    { key: 'week', label: 'Last 7 days' },
+    { key: 'earlier', label: 'Earlier' },
+  ]
+  return meta
+    .map(({ key, label }) => {
+      const items = withFailedPinned(buckets[key])
+      const failedCount = items.filter((e) => e.status === 'error').length
+      return { key, label, items, failedCount }
+    })
+    .filter((g) => g.items.length > 0)
+})
+
 watch(filterModel, async (m) => {
   await store.applyFilter({
     status: m.status,
@@ -120,4 +201,29 @@ async function openDetail(id) {
 .fc-dl__sentinel {
   display: flex; justify-content: center; padding: 1rem 0;
 }
+
+.fc-dl__group { margin-bottom: 12px; }
+.fc-dl__group-head {
+  display: flex; align-items: center; gap: 8px;
+  padding: 6px 8px;
+  background: rgb(var(--v-theme-on-surface) / 0.04);
+  border-radius: 4px;
+  cursor: pointer;
+  user-select: none;
+}
+.fc-dl__group-head:hover {
+  background: rgb(var(--v-theme-on-surface) / 0.08);
+}
+.fc-dl__group-chev {
+  color: rgb(var(--v-theme-on-surface-variant));
+}
+.fc-dl__group-label {
+  font-weight: 600;
+  color: rgb(var(--v-theme-on-surface));
+  flex: 1;
+}
+.fc-dl__group-counts {
+  display: flex; gap: 6px; align-items: center;
+}
+.fc-dl__group-body { margin-top: 4px; }
 </style>

From 407de18ff61b687a7658adf7aac82c661714ae00 Mon Sep 17 00:00:00 2001
From: Bryan Van Deusen <bryan.vandeusen@fabledsword.com>
Date: Wed, 27 May 2026 22:23:35 -0400
Subject: [PATCH 2/7] fix(ml): video branch needs longer time limits; recovery
 sweep is now per-queue
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Operator-flagged 2026-05-28: tag_and_embed on image 6288 (an mp4) was
marked failed by recover_stalled_task_runs at the 5-min sweep tick
while still legitimately running. The error_type='RecoverySweep' /
"no completion signal received within 5 min" message was misleading
— the worker was busy, not stuck.

Root cause is two interacting limits, both undersized for video work:

  tag_and_embed: soft_time_limit=300, time_limit=420
                 (sized for the image branch, ≈2 GPU ops)
  recovery sweep: STUCK_THRESHOLD_MINUTES = 5 across all queues

The video branch samples 10 frames via ffmpeg, then runs tagger +
embedder on EACH frame — ~20 GPU ops vs 2 for an image. A loaded
ml-worker can take 5-10 min on a long video, which trips both
limits well before the task naturally finishes.

**Two-part fix**

1. `tag_and_embed` time limits bumped to soft=900 (15 min) / time=1200
   (20 min). Sized for the video path's worst case; image runs return
   in seconds and don't care.

2. New `QUEUE_STUCK_THRESHOLD_MINUTES` override dict in maintenance.py.
   Queues with legitimately-long-running tasks (currently just `ml` at
   25 min — 5-min buffer past the new hard kill) get their own
   threshold; queues not in the dict use the default 5 min. The sweep
   now issues one UPDATE per distinct threshold value, with
   `queue.notin_(override_queues)` on the default pass so each row is
   touched at most once.

Tests:
- _make_task_run helper accepts `queue=` (defaults to "default") so
  existing tests use the default-threshold path.
- New test `test_recover_stalled_task_runs_ml_queue_uses_longer_threshold`
  pins both directions: a 10-min-old ml row survives (fresh by 25-min
  override), a 30-min-old ml row gets flagged.

After deploy, operator's mp4 ML jobs run to completion without
spurious RecoverySweep failures.
---
 backend/app/tasks/maintenance.py | 73 ++++++++++++++++++++++++++++----
 backend/app/tasks/ml.py          | 11 ++++-
 tests/test_maintenance.py        | 45 +++++++++++++++++++-
 3 files changed, 116 insertions(+), 13 deletions(-)

diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py
index df854a3..c5bb617 100644
--- a/backend/app/tasks/maintenance.py
+++ b/backend/app/tasks/maintenance.py
@@ -24,6 +24,21 @@ FFPROBE_TIMEOUT_SECONDS = 10
 TASK_RUN_KEEP_OK_SECONDS = 24 * 3600       # 24 h
 TASK_RUN_KEEP_FAILURE_SECONDS = 7 * 24 * 3600  # 7 days
 
+# Per-queue overrides for recover_stalled_task_runs. Queues whose
+# tasks can legitimately run longer than the default 5-min threshold
+# need their own larger value, otherwise the sweep marks in-flight
+# tasks 'error' before they get a chance to finish. The dict's value
+# MUST be ≥ the longest task.time_limit on the queue + a small buffer.
+#
+# ml: tag_and_embed video branch samples 10 frames, runs tagger +
+#     embedder on each — soft_time_limit=900 / time_limit=1200; sweep
+#     at 25 min gives a 5-min buffer past the hard kill.
+#     Operator-flagged 2026-05-28 (image 6288, an mp4, marked failed
+#     at the 5-min sweep tick while still processing).
+QUEUE_STUCK_THRESHOLD_MINUTES: dict[str, int] = {
+    "ml": 25,
+}
+
 
 @celery.task(name="backend.app.tasks.maintenance.recover_interrupted_tasks")
 def recover_interrupted_tasks() -> int:
@@ -121,21 +136,35 @@ def cleanup_old_tasks() -> int:
 
 @celery.task(name="backend.app.tasks.maintenance.recover_stalled_task_runs")
 def recover_stalled_task_runs() -> int:
-    """Flip task_run rows stuck in 'running' for >STUCK_THRESHOLD_MINUTES
-    to 'error'. FC-3i.
+    """Flip task_run rows stuck in 'running' past their queue-specific
+    threshold to 'error'. FC-3i.
 
     A row gets stuck when the worker dies without emitting
     task_postrun / task_failure (e.g. OOM, container restart between
-    signals, signal handler raised+logged). Shares the 5-min threshold
-    with recover_interrupted_tasks for consistency.
+    signals, signal handler raised+logged). The default 5-min threshold
+    fits short-lived queues (import/thumbnail/download); queues that
+    legitimately run longer tasks (ml-video, deep scans) get their
+    own larger threshold via QUEUE_STUCK_THRESHOLD_MINUTES so the
+    sweep doesn't preempt them.
+
+    Runs once per distinct threshold value: each pass updates rows
+    whose queue maps to that threshold.
     """
     SessionLocal = _sync_session_factory()
-    cutoff = datetime.now(UTC) - timedelta(minutes=STUCK_THRESHOLD_MINUTES)
+    now = datetime.now(UTC)
+    # Group queues by their threshold value so we issue one UPDATE
+    # per distinct threshold. Queues NOT in the override dict use the
+    # default; their UPDATE excludes the override queues so each row
+    # is touched at most once.
+    override_queues = set(QUEUE_STUCK_THRESHOLD_MINUTES.keys())
+    total = 0
     with SessionLocal() as session:
-        result = session.execute(
+        # Default-threshold pass — all queues except the overridden ones.
+        default_cutoff = now - timedelta(minutes=STUCK_THRESHOLD_MINUTES)
+        default_stmt = (
             update(TaskRun)
             .where(TaskRun.status == "running")
-            .where(TaskRun.started_at < cutoff)
+            .where(TaskRun.started_at < default_cutoff)
             .values(
                 status="error",
                 error_type="RecoverySweep",
@@ -143,11 +172,37 @@ def recover_stalled_task_runs() -> int:
                     f"no completion signal received within "
                     f"{STUCK_THRESHOLD_MINUTES} min"
                 ),
-                finished_at=datetime.now(UTC),
+                finished_at=now,
             )
         )
+        if override_queues:
+            default_stmt = default_stmt.where(
+                TaskRun.queue.notin_(override_queues)
+            )
+        total += session.execute(default_stmt).rowcount or 0
+
+        # Per-queue override passes.
+        for queue, minutes in QUEUE_STUCK_THRESHOLD_MINUTES.items():
+            cutoff = now - timedelta(minutes=minutes)
+            stmt = (
+                update(TaskRun)
+                .where(TaskRun.status == "running")
+                .where(TaskRun.queue == queue)
+                .where(TaskRun.started_at < cutoff)
+                .values(
+                    status="error",
+                    error_type="RecoverySweep",
+                    error_message=(
+                        f"no completion signal received within "
+                        f"{minutes} min"
+                    ),
+                    finished_at=now,
+                )
+            )
+            total += session.execute(stmt).rowcount or 0
+
         session.commit()
-        return result.rowcount or 0
+        return total
 
 
 @celery.task(name="backend.app.tasks.maintenance.prune_task_runs")
diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py
index 75de41d..c4b8ddc 100644
--- a/backend/app/tasks/ml.py
+++ b/backend/app/tasks/ml.py
@@ -31,8 +31,15 @@ def _is_video(path: Path) -> bool:
     retry_backoff_max=60,
     retry_jitter=True,
     max_retries=3,
-    soft_time_limit=300,
-    time_limit=420,
+    # Sized for the video branch: sample 10 frames, run tagger +
+    # embedder on each (≈20 GPU ops vs 2 for an image). A loaded
+    # ml-worker can take 5-10 min on a long video; bumped from
+    # 5min/7min on 2026-05-28 after operator-flagged image 6288 (a
+    # .mp4) hit the recovery sweep at 5 min while still legitimately
+    # processing. Image runs return in seconds; the bump doesn't
+    # affect their UX.
+    soft_time_limit=900,   # 15 min
+    time_limit=1200,       # 20 min hard
 )
 def tag_and_embed(self, image_id: int) -> dict:
     """Run Camie + SigLIP on one image; store predictions + embedding;
diff --git a/tests/test_maintenance.py b/tests/test_maintenance.py
index 7fc04e9..9f2803e 100644
--- a/tests/test_maintenance.py
+++ b/tests/test_maintenance.py
@@ -195,11 +195,11 @@ def test_cleanup_old_deletes_finished_old(db_sync):
 
 
 def _make_task_run(db_sync, *, status, started_at, finished_at=None,
-                   error_type=None):
+                   error_type=None, queue="default"):
     from backend.app.models import TaskRun
     row = TaskRun(
         celery_task_id="x",
-        queue="ml",
+        queue=queue,
         task_name="backend.app.tasks.fake.t",
         target_id=1,
         started_at=started_at,
@@ -255,6 +255,47 @@ def test_recover_stalled_task_runs_skips_fresh_running(db_sync):
     recovered = recover_stalled_task_runs.apply().get()
     assert recovered == 0
 
+
+def test_recover_stalled_task_runs_ml_queue_uses_longer_threshold(db_sync):
+    """ml-queue tasks (tag_and_embed video branch) legitimately run
+    past the default 5-min threshold. The sweep must NOT flag an
+    ml-queue task that's only been running 10 min — the override
+    threshold (25 min via QUEUE_STUCK_THRESHOLD_MINUTES) protects
+    in-flight video tagging. Operator-flagged 2026-05-28 after
+    image 6288 (mp4) was marked failed at the 5-min tick mid-run."""
+    from sqlalchemy import select
+
+    from backend.app.models import TaskRun
+    from backend.app.tasks.maintenance import recover_stalled_task_runs
+
+    now = datetime.now(UTC)
+    # 10-min-old ml-queue row: stale by the default 5-min rule but
+    # fresh by the 25-min ml override. Must survive the sweep.
+    ml_fresh_id = _make_task_run(
+        db_sync, status="running", queue="ml",
+        started_at=now - timedelta(minutes=10),
+    )
+    # 30-min-old ml-queue row: past even the ml override. Must be
+    # flagged.
+    ml_stale_id = _make_task_run(
+        db_sync, status="running", queue="ml",
+        started_at=now - timedelta(minutes=30),
+    )
+    db_sync.commit()
+
+    recovered = recover_stalled_task_runs.apply().get()
+    assert recovered == 1
+
+    db_sync.expire_all()
+    ml_fresh_status = db_sync.execute(
+        select(TaskRun.status).where(TaskRun.id == ml_fresh_id)
+    ).scalar_one()
+    ml_stale_status = db_sync.execute(
+        select(TaskRun.status).where(TaskRun.id == ml_stale_id)
+    ).scalar_one()
+    assert ml_fresh_status == "running"
+    assert ml_stale_status == "error"
+
     db_sync.expire_all()
     status = db_sync.execute(
         select(TaskRun.status).where(TaskRun.id == fresh_id)

From a85880f965387c1ed3156a38055fa023a921447a Mon Sep 17 00:00:00 2001
From: Bryan Van Deusen <bryan.vandeusen@fabledsword.com>
Date: Wed, 27 May 2026 22:45:11 -0400
Subject: [PATCH 3/7] fix(import): split archive imports into their own task +
 budget; archive-aware recovery sweeps
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Operator-flagged 2026-05-28: import_media_file on target 1645019 hit
SoftTimeLimitExceeded at exactly 5.0 min. Their diagnosis was correct —
the timeout covered the WHOLE archive, not per object. Importer._import_archive
(importer.py:409) runs the full per-member pipeline (sha256 + pHash +
dedup query + copy + provenance) for EVERY media member inline, all
under import_media_file's single 300s soft limit. A single media file
is sub-second; a multi-hundred-member archive blows the budget. They
shared one task name and one timeout.

**Split archive into its own task**

- New `import_archive_file` task: same body as import_media_file
  (dispatch is by file-kind inside Importer.import_one) but
  soft=30min / hard=35min. Shared `_run_import_task` helper holds the
  flip-to-processing + resilience-contract wrapper; both tasks call it.
- New `enqueue_import(task_id, task_type)` router — single source of
  truth for media-vs-archive dispatch. Used by all three enqueue sites:
  scan_directory, /api/import/retry-failed, recover_interrupted_tasks.
- scan_directory now sets ImportTask.task_type = "archive" when
  is_archive(entry) (the model field already existed, anticipating
  this; scan was hardcoding "media").
- import_archive_file routes to the existing 'import' queue via the
  task_routes `import_file.*` wildcard — no worker config change.

**Archive-aware recovery sweeps**

Both sweeps would otherwise preempt a legitimately-running archive:

- recover_interrupted_tasks (ImportTask 'processing' sweep): now
  task-type-aware. Media stays at STUCK_THRESHOLD_MINUTES (5); archives
  get ARCHIVE_STUCK_THRESHOLD_MINUTES (40 = 5-min buffer past the
  35-min hard limit). Single UPDATE with an OR predicate over the two
  (task_type, cutoff) pairs; requeue routes via enqueue_import.
- recover_stalled_task_runs (TaskRun 'running' sweep): now supports
  per-task-name overrides (TASK_STUCK_THRESHOLD_MINUTES) layered above
  the per-queue overrides added for ml. import_archive_file gets 40 min
  while the 'import' queue stays at the 5-min default for single-file
  imports. Precedence: task_name → queue → default, each pass excluding
  rows claimed by a higher-precedence pass so every row is touched once.

**Tests**

- test_import_archive_file_registered
- test_recover_stalled_task_runs_archive_task_uses_longer_threshold —
  pins that a 10-min archive task-run survives, a 50-min one is flagged,
  and a same-queue 10-min media import is flagged at the default.
- _make_task_run gains queue= + task_name= params.

After deploy: archive imports get a 30-min budget and aren't preempted
by either sweep; single-file imports keep their tight 5-min detection.
---
 backend/app/api/import_admin.py  |  14 +--
 backend/app/tasks/import_file.py | 101 +++++++++++++++------
 backend/app/tasks/maintenance.py | 148 ++++++++++++++++++-------------
 backend/app/tasks/scan.py        |  14 +--
 tests/test_maintenance.py        |  52 ++++++++++-
 tests/test_tasks_register.py     |   4 +
 6 files changed, 232 insertions(+), 101 deletions(-)

diff --git a/backend/app/api/import_admin.py b/backend/app/api/import_admin.py
index 02f65d2..2ebeb9b 100644
--- a/backend/app/api/import_admin.py
+++ b/backend/app/api/import_admin.py
@@ -114,18 +114,18 @@ async def retry_failed():
                 status="queued", error=None,
                 started_at=None, finished_at=None,
             )
-            .returning(ImportTask.id)
+            .returning(ImportTask.id, ImportTask.task_type)
         )
-        failed_ids = [row[0] for row in result.all()]
-        if not failed_ids:
+        failed = result.all()
+        if not failed:
             return jsonify({"retried": 0})
         await session.commit()
 
-    from ..tasks.import_file import import_media_file
-    for tid in failed_ids:
-        import_media_file.delay(tid)
+    from ..tasks.import_file import enqueue_import
+    for tid, task_type in failed:
+        enqueue_import(tid, task_type)
 
-    return jsonify({"retried": len(failed_ids)})
+    return jsonify({"retried": len(failed)})
 
 
 @import_admin_bp.route("/clear-stuck", methods=["POST"])
diff --git a/backend/app/tasks/import_file.py b/backend/app/tasks/import_file.py
index e0752e8..4a124d1 100644
--- a/backend/app/tasks/import_file.py
+++ b/backend/app/tasks/import_file.py
@@ -64,30 +64,13 @@ def _mark_failed(session, task, error_msg: str) -> None:
             pass
 
 
-@celery.task(
-    name="backend.app.tasks.import_file.import_media_file",
-    bind=True,
-    autoretry_for=(OperationalError, DBAPIError, OSError),
-    retry_backoff=5,
-    retry_backoff_max=60,
-    retry_jitter=True,
-    max_retries=3,
-    soft_time_limit=300,
-    time_limit=360,
-)
-def import_media_file(self, import_task_id: int) -> dict:
-    """Returns a dict so the eager-mode tests can assert without DB.
-
-    Decorator notes:
-      - autoretry_for: transient DB / filesystem errors retry with
-        exponential backoff (5s base, jitter, max 3 attempts). On final
-        give-up the task raises and acks_late=True (set globally on the
-        Celery app) does NOT redeliver — the recovery sweep catches the
-        row instead.
-      - soft_time_limit (300s) raises SoftTimeLimitExceeded in this
-        process so the task can mark its row failed before being killed.
-      - time_limit (360s) is the hard cap; SIGKILL if the soft signal
-        was swallowed.
+def _run_import_task(import_task_id: int) -> dict:
+    """Shared body for import_media_file + import_archive_file. The two
+    tasks differ ONLY in their Celery time limits (a single media file
+    is sub-second; an archive runs the full per-member pipeline inline
+    for every member and can take many minutes). Both flip the row to
+    'processing', dispatch to `_do_import`, and honor the
+    flip-to-terminal resilience contract.
     """
     SessionLocal = _sync_session_factory()
     with SessionLocal() as session:
@@ -103,19 +86,85 @@ def import_media_file(self, import_task_id: int) -> dict:
         try:
             return _do_import(session, task, import_task_id)
         except SoftTimeLimitExceeded:
-            _mark_failed(session, task, "soft_time_limit exceeded (>300s)")
+            _mark_failed(session, task, "soft_time_limit exceeded")
             raise
         except (OperationalError, DBAPIError, OSError):
             # Retryable per the decorator; do NOT mark failed (let
             # autoretry have a clean go at it). If autoretry exhausts,
             # the row stays 'processing' and the maintenance sweep
-            # flips it within 5 min.
+            # flips it.
             raise
         except Exception as exc:  # noqa: BLE001 — pipeline crash, mark + re-raise
             _mark_failed(session, task, f"{type(exc).__name__}: {exc}")
             raise
 
 
+@celery.task(
+    name="backend.app.tasks.import_file.import_media_file",
+    bind=True,
+    autoretry_for=(OperationalError, DBAPIError, OSError),
+    retry_backoff=5,
+    retry_backoff_max=60,
+    retry_jitter=True,
+    max_retries=3,
+    soft_time_limit=300,
+    time_limit=360,
+)
+def import_media_file(self, import_task_id: int) -> dict:
+    """Import ONE media file (or non-media → PostAttachment). Sub-second
+    for the common case; the tight 5-min soft limit keeps a genuinely
+    stuck single-file import detectable fast.
+
+    Decorator notes:
+      - autoretry_for: transient DB / filesystem errors retry with
+        exponential backoff (5s base, jitter, max 3 attempts). On final
+        give-up the task raises and acks_late=True (set globally on the
+        Celery app) does NOT redeliver — the recovery sweep catches the
+        row instead.
+      - soft_time_limit (300s) raises SoftTimeLimitExceeded in-process
+        so the task can mark its row failed before being killed.
+      - time_limit (360s) is the hard SIGKILL cap.
+    """
+    return _run_import_task(import_task_id)
+
+
+@celery.task(
+    name="backend.app.tasks.import_file.import_archive_file",
+    bind=True,
+    autoretry_for=(OperationalError, DBAPIError, OSError),
+    retry_backoff=5,
+    retry_backoff_max=60,
+    retry_jitter=True,
+    max_retries=3,
+    # Archives run the full per-member pipeline (sha256 + pHash + dedup
+    # query + copy + provenance) for EVERY media member inline, under a
+    # single task budget. A multi-hundred-member archive blows the
+    # 5-min media limit. soft=30min / hard=35min sizes for a large
+    # archive. Operator-flagged 2026-05-28 (target 1645019 hit the old
+    # shared 300s soft limit). The recovery sweep gives this task its
+    # own 40-min threshold via maintenance.TASK_STUCK_THRESHOLD_MINUTES
+    # so it isn't preempted while legitimately grinding through members.
+    soft_time_limit=1800,
+    time_limit=2100,
+)
+def import_archive_file(self, import_task_id: int) -> dict:
+    """Import an archive: extract + run the per-member media pipeline for
+    every member inline, then preserve the archive as a PostAttachment.
+    Same body as import_media_file (dispatch is by file kind inside
+    Importer.import_one); split out purely for the larger time budget."""
+    return _run_import_task(import_task_id)
+
+
+def enqueue_import(task_id: int, task_type: str) -> None:
+    """Route an ImportTask to the right Celery task by its task_type.
+    Single source of truth for the media-vs-archive dispatch so the
+    scan, retry, and recovery-requeue paths stay in sync."""
+    if task_type == "archive":
+        import_archive_file.delay(task_id)
+    else:
+        import_media_file.delay(task_id)
+
+
 def _do_import(session, task, import_task_id: int) -> dict:
     """Actual work, called from inside the resilience wrapper."""
     settings = session.execute(
diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py
index c5bb617..797d6f2 100644
--- a/backend/app/tasks/maintenance.py
+++ b/backend/app/tasks/maintenance.py
@@ -16,6 +16,13 @@ from ._sync_engine import sync_session_factory as _sync_session_factory
 log = logging.getLogger(__name__)
 
 STUCK_THRESHOLD_MINUTES = 5
+# Archive ImportTasks run the per-member pipeline inline for every
+# member (import_archive_file: soft=30min/hard=35min). The ImportTask
+# 'processing' recovery sweep must give them a longer threshold or it
+# re-queues a legitimately-running archive mid-import (double-process).
+# 40 min = 5-min buffer past the archive task's hard kill.
+# Operator-flagged 2026-05-28 (target 1645019, a big archive).
+ARCHIVE_STUCK_THRESHOLD_MINUTES = 40
 ORPHAN_PENDING_THRESHOLD_MINUTES = 30
 OLD_TASK_DAYS = 7
 PHASH_PAGE = 500
@@ -24,32 +31,40 @@ FFPROBE_TIMEOUT_SECONDS = 10
 TASK_RUN_KEEP_OK_SECONDS = 24 * 3600       # 24 h
 TASK_RUN_KEEP_FAILURE_SECONDS = 7 * 24 * 3600  # 7 days
 
-# Per-queue overrides for recover_stalled_task_runs. Queues whose
-# tasks can legitimately run longer than the default 5-min threshold
-# need their own larger value, otherwise the sweep marks in-flight
-# tasks 'error' before they get a chance to finish. The dict's value
-# MUST be ≥ the longest task.time_limit on the queue + a small buffer.
+# Overrides for recover_stalled_task_runs (the TaskRun 'running' sweep).
+# Tasks/queues that legitimately run longer than the default 5-min
+# threshold need their own larger value, else the sweep marks in-flight
+# work 'error' before it finishes. Each value MUST be ≥ the relevant
+# task.time_limit + a small buffer. task_name overrides take precedence
+# over queue overrides.
 #
-# ml: tag_and_embed video branch samples 10 frames, runs tagger +
-#     embedder on each — soft_time_limit=900 / time_limit=1200; sweep
-#     at 25 min gives a 5-min buffer past the hard kill.
-#     Operator-flagged 2026-05-28 (image 6288, an mp4, marked failed
-#     at the 5-min sweep tick while still processing).
+# ml queue: tag_and_embed video branch (≈20 GPU ops); time_limit=1200.
+# import_archive_file: shares the 'import' queue with the fast
+#   single-file import_media_file, so it needs a task-name override
+#   (the import queue itself stays at the 5-min default for single
+#   files); time_limit=2100.
 QUEUE_STUCK_THRESHOLD_MINUTES: dict[str, int] = {
     "ml": 25,
 }
+TASK_STUCK_THRESHOLD_MINUTES: dict[str, int] = {
+    "backend.app.tasks.import_file.import_archive_file": 40,
+}
 
 
 @celery.task(name="backend.app.tasks.maintenance.recover_interrupted_tasks")
 def recover_interrupted_tasks() -> int:
     """Recover stuck ImportTask rows. Two distinct stuck states:
 
-    1. 'processing' > 5 min — worker crash mid-import. Re-queue via
-       .delay() and let the import retry. Was 30 min historically;
-       tightened 2026-05-24 after operator hit a 2224-row zombie pile.
-       import_media_file is sub-second for the vast majority of files and
-       capped at the per-task soft_time_limit (5 min), so anything still
-       'processing' after that window is a confirmed crash.
+    1. 'processing' too long — worker crash mid-import. Re-queue via
+       enqueue_import (routing media vs archive) and let the import
+       retry. Threshold is task-type-aware: media files are sub-second
+       and capped at the 5-min soft limit, so STUCK_THRESHOLD_MINUTES
+       (5) means a confirmed crash; archives run the per-member
+       pipeline inline (import_archive_file, 35-min hard limit) so they
+       get ARCHIVE_STUCK_THRESHOLD_MINUTES (40) to avoid re-queueing a
+       still-running archive. (Media was tightened from 30 min to 5
+       2026-05-24 after a 2224-row zombie pile; archive split out
+       2026-05-28.)
 
     2. 'pending' or 'queued' > 30 min — enqueue-phase crash. scan_directory
        creates rows with status='pending' (commit), then in a second pass
@@ -66,7 +81,8 @@ def recover_interrupted_tasks() -> int:
     """
     SessionLocal = _sync_session_factory()
     now = datetime.now(UTC)
-    processing_cutoff = now - timedelta(minutes=STUCK_THRESHOLD_MINUTES)
+    media_cutoff = now - timedelta(minutes=STUCK_THRESHOLD_MINUTES)
+    archive_cutoff = now - timedelta(minutes=ARCHIVE_STUCK_THRESHOLD_MINUTES)
     orphan_cutoff = now - timedelta(minutes=ORPHAN_PENDING_THRESHOLD_MINUTES)
     with SessionLocal() as session:
         # Both sweeps used to be SELECT ids → UPDATE WHERE id IN (...) which
@@ -74,20 +90,31 @@ def recover_interrupted_tasks() -> int:
         # tens of thousands of rows (operator hit it 2026-05-26 after the
         # /import deep scan piled up orphans). Folding the SELECT into the
         # UPDATE eliminates the IN-list entirely. RETURNING gives us back
-        # exactly the ids that flipped so the stuck sweep can still
-        # .delay() each one.
+        # exactly the (id, task_type) pairs that flipped so the requeue
+        # can route media vs archive correctly.
+        #
+        # Media + archive get separate cutoffs: a single media file is
+        # sub-second so 5 min means crash; an archive runs the per-member
+        # pipeline inline and can legitimately take up to its 35-min hard
+        # limit, so it gets ARCHIVE_STUCK_THRESHOLD_MINUTES (40) to avoid
+        # re-queueing a still-running archive.
         stuck_result = session.execute(
             update(ImportTask)
             .where(ImportTask.status == "processing")
-            .where(ImportTask.started_at < processing_cutoff)
+            .where(
+                ((ImportTask.task_type != "archive")
+                 & (ImportTask.started_at < media_cutoff))
+                | ((ImportTask.task_type == "archive")
+                   & (ImportTask.started_at < archive_cutoff))
+            )
             .values(
                 status="queued",
                 started_at=None,
                 error="recovered from stuck state",
             )
-            .returning(ImportTask.id)
+            .returning(ImportTask.id, ImportTask.task_type)
         )
-        stuck_ids = [row[0] for row in stuck_result.all()]
+        stuck = stuck_result.all()
 
         orphan_result = session.execute(
             update(ImportTask)
@@ -106,12 +133,12 @@ def recover_interrupted_tasks() -> int:
 
         session.commit()
 
-        if stuck_ids:
-            from .import_file import import_media_file
-            for tid in stuck_ids:
-                import_media_file.delay(tid)
+        if stuck:
+            from .import_file import enqueue_import
+            for tid, task_type in stuck:
+                enqueue_import(tid, task_type)
 
-        return len(stuck_ids) + orphan_count
+        return len(stuck) + orphan_count
 
 
 @celery.task(name="backend.app.tasks.maintenance.cleanup_old_tasks")
@@ -152,54 +179,53 @@ def recover_stalled_task_runs() -> int:
     """
     SessionLocal = _sync_session_factory()
     now = datetime.now(UTC)
-    # Group queues by their threshold value so we issue one UPDATE
-    # per distinct threshold. Queues NOT in the override dict use the
-    # default; their UPDATE excludes the override queues so each row
-    # is touched at most once.
+    override_tasks = set(TASK_STUCK_THRESHOLD_MINUTES.keys())
     override_queues = set(QUEUE_STUCK_THRESHOLD_MINUTES.keys())
     total = 0
-    with SessionLocal() as session:
-        # Default-threshold pass — all queues except the overridden ones.
-        default_cutoff = now - timedelta(minutes=STUCK_THRESHOLD_MINUTES)
-        default_stmt = (
+
+    def _flag(minutes, *extra_where):
+        cutoff = now - timedelta(minutes=minutes)
+        stmt = (
             update(TaskRun)
             .where(TaskRun.status == "running")
-            .where(TaskRun.started_at < default_cutoff)
+            .where(TaskRun.started_at < cutoff)
             .values(
                 status="error",
                 error_type="RecoverySweep",
                 error_message=(
-                    f"no completion signal received within "
-                    f"{STUCK_THRESHOLD_MINUTES} min"
+                    f"no completion signal received within {minutes} min"
                 ),
                 finished_at=now,
             )
         )
-        if override_queues:
-            default_stmt = default_stmt.where(
-                TaskRun.queue.notin_(override_queues)
-            )
-        total += session.execute(default_stmt).rowcount or 0
+        for w in extra_where:
+            stmt = stmt.where(w)
+        return session.execute(stmt).rowcount or 0
 
-        # Per-queue override passes.
+    with SessionLocal() as session:
+        # Precedence: task_name override → queue override → default.
+        # Each pass excludes rows claimed by a higher-precedence pass so
+        # every row is touched at most once.
+
+        # 1. Per-task-name overrides (e.g. import_archive_file, which
+        #    shares the 'import' queue with fast single-file imports).
+        for task_name, minutes in TASK_STUCK_THRESHOLD_MINUTES.items():
+            total += _flag(minutes, TaskRun.task_name == task_name)
+
+        # 2. Per-queue overrides, excluding the override task-names.
         for queue, minutes in QUEUE_STUCK_THRESHOLD_MINUTES.items():
-            cutoff = now - timedelta(minutes=minutes)
-            stmt = (
-                update(TaskRun)
-                .where(TaskRun.status == "running")
-                .where(TaskRun.queue == queue)
-                .where(TaskRun.started_at < cutoff)
-                .values(
-                    status="error",
-                    error_type="RecoverySweep",
-                    error_message=(
-                        f"no completion signal received within "
-                        f"{minutes} min"
-                    ),
-                    finished_at=now,
-                )
-            )
-            total += session.execute(stmt).rowcount or 0
+            wheres = [TaskRun.queue == queue]
+            if override_tasks:
+                wheres.append(TaskRun.task_name.notin_(override_tasks))
+            total += _flag(minutes, *wheres)
+
+        # 3. Default — everything not claimed above.
+        default_wheres = []
+        if override_queues:
+            default_wheres.append(TaskRun.queue.notin_(override_queues))
+        if override_tasks:
+            default_wheres.append(TaskRun.task_name.notin_(override_tasks))
+        total += _flag(STUCK_THRESHOLD_MINUTES, *default_wheres)
 
         session.commit()
         return total
diff --git a/backend/app/tasks/scan.py b/backend/app/tasks/scan.py
index 2e5b612..0451e49 100644
--- a/backend/app/tasks/scan.py
+++ b/backend/app/tasks/scan.py
@@ -17,6 +17,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_asyn
 from ..celery_app import celery
 from ..config import get_config
 from ..models import DownloadEvent, ImportBatch, ImportSettings, ImportTask
+from ..services.archive_extractor import is_archive
 from ..services.scheduler_service import select_due_sources
 from ._sync_engine import sync_session_factory as _sync_session_factory
 
@@ -96,7 +97,9 @@ def scan_directory(self, triggered_by: str = "manual",
             task = ImportTask(
                 batch_id=batch_id,
                 source_path=entry_str,
-                task_type="media",
+                # Archives route to import_archive_file (larger time
+                # budget) — they run the per-member pipeline inline.
+                task_type="archive" if is_archive(entry) else "media",
                 status="pending",
                 size_bytes=size,
             )
@@ -115,15 +118,16 @@ def scan_directory(self, triggered_by: str = "manual",
             batch.finished_at = datetime.now(UTC)
         session.commit()
 
-        # Now enqueue import_media_file for each pending task.
+        # Now enqueue each pending task on the right Celery task
+        # (media vs archive) via the shared router.
+        from .import_file import enqueue_import
+
         for task in session.execute(
             select(ImportTask).where(ImportTask.batch_id == batch_id)
         ).scalars():
             task.status = "queued"
             session.add(task)
-            from .import_file import import_media_file
-
-            import_media_file.delay(task.id)
+            enqueue_import(task.id, task.task_type)
         session.commit()
 
         if mode == "deep":
diff --git a/tests/test_maintenance.py b/tests/test_maintenance.py
index 9f2803e..42d1871 100644
--- a/tests/test_maintenance.py
+++ b/tests/test_maintenance.py
@@ -195,12 +195,13 @@ def test_cleanup_old_deletes_finished_old(db_sync):
 
 
 def _make_task_run(db_sync, *, status, started_at, finished_at=None,
-                   error_type=None, queue="default"):
+                   error_type=None, queue="default",
+                   task_name="backend.app.tasks.fake.t"):
     from backend.app.models import TaskRun
     row = TaskRun(
         celery_task_id="x",
         queue=queue,
-        task_name="backend.app.tasks.fake.t",
+        task_name=task_name,
         target_id=1,
         started_at=started_at,
         finished_at=finished_at,
@@ -296,6 +297,53 @@ def test_recover_stalled_task_runs_ml_queue_uses_longer_threshold(db_sync):
     assert ml_fresh_status == "running"
     assert ml_stale_status == "error"
 
+
+def test_recover_stalled_task_runs_archive_task_uses_longer_threshold(db_sync):
+    """import_archive_file shares the 'import' queue with fast
+    single-file import_media_file, so it gets a per-task-name override
+    (40 min) while the import queue stays at the 5-min default. A
+    10-min-old archive task-run must survive; a 50-min-old one is
+    flagged. Operator-flagged 2026-05-28."""
+    from sqlalchemy import select
+
+    from backend.app.models import TaskRun
+    from backend.app.tasks.maintenance import recover_stalled_task_runs
+
+    archive_name = "backend.app.tasks.import_file.import_archive_file"
+    now = datetime.now(UTC)
+    # Fast single-file import on the same queue, 10 min old → flagged
+    # by the default 5-min rule.
+    media_id = _make_task_run(
+        db_sync, status="running", queue="import",
+        task_name="backend.app.tasks.import_file.import_media_file",
+        started_at=now - timedelta(minutes=10),
+    )
+    # Archive on the same queue, 10 min old → survives (40-min override).
+    archive_fresh_id = _make_task_run(
+        db_sync, status="running", queue="import",
+        task_name=archive_name,
+        started_at=now - timedelta(minutes=10),
+    )
+    # Archive 50 min old → past even the 40-min override → flagged.
+    archive_stale_id = _make_task_run(
+        db_sync, status="running", queue="import",
+        task_name=archive_name,
+        started_at=now - timedelta(minutes=50),
+    )
+    db_sync.commit()
+
+    recovered = recover_stalled_task_runs.apply().get()
+    assert recovered == 2  # media + stale archive
+
+    db_sync.expire_all()
+    def _status(_id):
+        return db_sync.execute(
+            select(TaskRun.status).where(TaskRun.id == _id)
+        ).scalar_one()
+    assert _status(media_id) == "error"
+    assert _status(archive_fresh_id) == "running"
+    assert _status(archive_stale_id) == "error"
+
     db_sync.expire_all()
     status = db_sync.execute(
         select(TaskRun.status).where(TaskRun.id == fresh_id)
diff --git a/tests/test_tasks_register.py b/tests/test_tasks_register.py
index 11f457c..2e5414f 100644
--- a/tests/test_tasks_register.py
+++ b/tests/test_tasks_register.py
@@ -21,6 +21,10 @@ def test_import_media_file_registered():
     assert "backend.app.tasks.import_file.import_media_file" in celery.tasks
 
 
+def test_import_archive_file_registered():
+    assert "backend.app.tasks.import_file.import_archive_file" in celery.tasks
+
+
 def test_generate_thumbnail_registered():
     assert "backend.app.tasks.thumbnail.generate_thumbnail" in celery.tasks
 

From 57a22d609838a5785f6038382a70737f946c4bff Mon Sep 17 00:00:00 2001
From: Bryan Van Deusen <bryan.vandeusen@fabledsword.com>
Date: Wed, 27 May 2026 23:06:00 -0400
Subject: [PATCH 4/7] =?UTF-8?q?fix(tests):=20repair=20test=5Fmaintenance?=
 =?UTF-8?q?=20=E2=80=94=20skips=5Ffresh=5Frunning=20tail=20was=20orphaned?=
 =?UTF-8?q?=20at=20module=20scope=20by=20the=20inserted=20ml/archive=20swe?=
 =?UTF-8?q?ep=20tests=20(F841/F821)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 tests/test_maintenance.py | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

diff --git a/tests/test_maintenance.py b/tests/test_maintenance.py
index 42d1871..0488943 100644
--- a/tests/test_maintenance.py
+++ b/tests/test_maintenance.py
@@ -256,6 +256,12 @@ def test_recover_stalled_task_runs_skips_fresh_running(db_sync):
     recovered = recover_stalled_task_runs.apply().get()
     assert recovered == 0
 
+    db_sync.expire_all()
+    status = db_sync.execute(
+        select(TaskRun.status).where(TaskRun.id == fresh_id)
+    ).scalar_one()
+    assert status == "running"
+
 
 def test_recover_stalled_task_runs_ml_queue_uses_longer_threshold(db_sync):
     """ml-queue tasks (tag_and_embed video branch) legitimately run
@@ -344,12 +350,6 @@ def test_recover_stalled_task_runs_archive_task_uses_longer_threshold(db_sync):
     assert _status(archive_fresh_id) == "running"
     assert _status(archive_stale_id) == "error"
 
-    db_sync.expire_all()
-    status = db_sync.execute(
-        select(TaskRun.status).where(TaskRun.id == fresh_id)
-    ).scalar_one()
-    assert status == "running"
-
 
 def test_prune_task_runs_deletes_ok_older_than_24h(db_sync):
     from sqlalchemy import select

From e77afe82952a7d5f25b19944992efa3c5a81639c Mon Sep 17 00:00:00 2001
From: Bryan Van Deusen <bryan.vandeusen@fabledsword.com>
Date: Wed, 27 May 2026 23:54:35 -0400
Subject: [PATCH 5/7] =?UTF-8?q?feat(import-resilience=20L1):=20poison-pill?=
 =?UTF-8?q?=20circuit=20breaker=20=E2=80=94=20cap=20stuck-task=20re-queues?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Layer 1 of the import-task resilience work (operator-requested
2026-05-28). The recover_interrupted_tasks sweep re-queues rows stuck
in 'processing' — correct for a worker crash, but without a cap a row
that RELIABLY hard-crashes the worker (OOM/segfault/SIGKILL on a
corrupt or oversized input) loops forever: re-queue → crash → re-queue,
burning a worker slot every 5 min. A caught exception flips to terminal
'failed' and never enters this loop; only process-killing inputs do.

- alembic 0026: import_task.recovery_count (int, default 0) +
  import_task.refetched (bool, default false — backs Layer 2).
- recover_interrupted_tasks now runs a poison-pill UPDATE FIRST: stuck
  rows whose recovery_count has already reached MAX_RECOVERY_ATTEMPTS-1
  are marked 'failed' with a diagnostic ("crashed or stalled the worker
  N times … likely a corrupt or oversized input … inspect/replace the
  file, then retry via /api/import/retry-failed") instead of re-queued.
  The re-queue pass then handles the remaining stuck rows and bumps
  recovery_count. Shared stuck_predicate (and_/or_) keeps the
  media-5min / archive-40min split.
- MAX_RECOVERY_ATTEMPTS=3 (two recoveries then give up).

The failed poison pill surfaces in the existing import-failures view
with its file path, directly answering "help me identify them."

Test test_recover_interrupted_poison_pill_caps_at_max pins both
branches: a row at the cap is failed (not re-enqueued, diagnostic
present), a row one short is re-queued + incremented.
---
 ...26_import_task_recovery_count_refetched.py | 53 ++++++++++++++++
 backend/app/models/import_task.py             | 18 +++++-
 backend/app/tasks/maintenance.py              | 63 ++++++++++++++++---
 tests/test_maintenance.py                     | 46 ++++++++++++++
 4 files changed, 170 insertions(+), 10 deletions(-)
 create mode 100644 alembic/versions/0026_import_task_recovery_count_refetched.py

diff --git a/alembic/versions/0026_import_task_recovery_count_refetched.py b/alembic/versions/0026_import_task_recovery_count_refetched.py
new file mode 100644
index 0000000..ccbc3da
--- /dev/null
+++ b/alembic/versions/0026_import_task_recovery_count_refetched.py
@@ -0,0 +1,53 @@
+"""import_task.recovery_count + refetched — poison-pill circuit breaker
+
+Revision ID: 0026
+Revises: 0025
+Create Date: 2026-05-28
+
+Backs the import-task resilience work (operator-flagged 2026-05-28):
+
+- recovery_count: how many times recover_interrupted_tasks has
+  re-queued this row from a stuck 'processing' state. A row that
+  hard-crashes the worker (OOM / segfault on a corrupt or oversized
+  input) leaves no terminal flip, so the sweep re-queues it — and
+  without a cap it would loop forever, re-crashing the worker each
+  time. After MAX_RECOVERY_ATTEMPTS the sweep marks it 'failed' with a
+  diagnostic instead.
+
+- refetched: whether a one-shot re-download has already been attempted
+  for this task's file. Bounds the Layer-2 re-fetch remediation to a
+  single attempt so source-side corruption doesn't loop.
+
+Both default to 0 / false; additive, no backfill needed.
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0026"
+down_revision: Union[str, None] = "0025"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "import_task",
+        sa.Column(
+            "recovery_count", sa.Integer(), nullable=False,
+            server_default="0",
+        ),
+    )
+    op.add_column(
+        "import_task",
+        sa.Column(
+            "refetched", sa.Boolean(), nullable=False,
+            server_default=sa.false(),
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("import_task", "refetched")
+    op.drop_column("import_task", "recovery_count")
diff --git a/backend/app/models/import_task.py b/backend/app/models/import_task.py
index da3d4ec..3c947c1 100644
--- a/backend/app/models/import_task.py
+++ b/backend/app/models/import_task.py
@@ -8,7 +8,16 @@ been processing longer than the stuck-task threshold.
 
 from datetime import datetime
 
-from sqlalchemy import BigInteger, DateTime, ForeignKey, Integer, String, Text, func
+from sqlalchemy import (
+    BigInteger,
+    Boolean,
+    DateTime,
+    ForeignKey,
+    Integer,
+    String,
+    Text,
+    func,
+)
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from .base import Base
@@ -26,6 +35,13 @@ class ImportTask(Base):
     task_type: Mapped[str] = mapped_column(String(16), nullable=False)  # media|archive
     status: Mapped[str] = mapped_column(String(16), nullable=False, default="pending", index=True)
 
+    # Poison-pill circuit breaker (alembic 0026). recovery_count tracks
+    # how many times the stuck-task sweep has re-queued this row; after
+    # the cap it's failed with a diagnostic instead of looping. refetched
+    # bounds the one-shot re-download remediation to a single attempt.
+    recovery_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    refetched: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+
     result_image_id: Mapped[int | None] = mapped_column(
         ForeignKey("image_record.id", ondelete="SET NULL"), nullable=True
     )
diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py
index 797d6f2..f864873 100644
--- a/backend/app/tasks/maintenance.py
+++ b/backend/app/tasks/maintenance.py
@@ -6,7 +6,7 @@ from datetime import UTC, datetime, timedelta
 from pathlib import Path
 
 from PIL import Image
-from sqlalchemy import delete, select, update
+from sqlalchemy import and_, delete, or_, select, update
 
 from ..celery_app import celery
 from ..models import DownloadEvent, ImageRecord, ImportSettings, ImportTask, TaskRun
@@ -23,6 +23,15 @@ STUCK_THRESHOLD_MINUTES = 5
 # 40 min = 5-min buffer past the archive task's hard kill.
 # Operator-flagged 2026-05-28 (target 1645019, a big archive).
 ARCHIVE_STUCK_THRESHOLD_MINUTES = 40
+
+# Poison-pill cap. After being recovered (re-queued from a stuck
+# 'processing' state) MAX_RECOVERY_ATTEMPTS-1 times, the next sweep
+# marks the row 'failed' instead of looping. 3 = two recoveries then
+# give up. A row reaches this only if it leaves NO terminal flip each
+# run — i.e. it hard-crashes the worker (OOM/segfault/SIGKILL), the
+# signature of a corrupt or oversized input. Caught exceptions already
+# flip to terminal 'failed' and never enter this loop.
+MAX_RECOVERY_ATTEMPTS = 3
 ORPHAN_PENDING_THRESHOLD_MINUTES = 30
 OLD_TASK_DAYS = 7
 PHASH_PAGE = 500
@@ -98,18 +107,54 @@ def recover_interrupted_tasks() -> int:
         # pipeline inline and can legitimately take up to its 35-min hard
         # limit, so it gets ARCHIVE_STUCK_THRESHOLD_MINUTES (40) to avoid
         # re-queueing a still-running archive.
+        stuck_predicate = and_(
+            ImportTask.status == "processing",
+            or_(
+                and_(ImportTask.task_type != "archive",
+                     ImportTask.started_at < media_cutoff),
+                and_(ImportTask.task_type == "archive",
+                     ImportTask.started_at < archive_cutoff),
+            ),
+        )
+
+        # POISON-PILL CIRCUIT BREAKER (Layer 1, 2026-05-28). A row that
+        # leaves no terminal flip (hard worker crash: OOM/segfault/SIGKILL
+        # on a corrupt or oversized input) gets re-queued by this sweep —
+        # and would loop forever, re-crashing the worker each pass,
+        # without a cap. Once a row has already been recovered
+        # MAX_RECOVERY_ATTEMPTS-1 times, stop re-queueing it and mark it
+        # 'failed' with a diagnostic so the operator can find + replace
+        # the offending file. This UPDATE runs FIRST so the rows it
+        # claims drop out of 'processing' before the re-queue pass.
+        poison_result = session.execute(
+            update(ImportTask)
+            .where(stuck_predicate)
+            .where(ImportTask.recovery_count >= MAX_RECOVERY_ATTEMPTS - 1)
+            .values(
+                status="failed",
+                finished_at=now,
+                error=(
+                    f"crashed or stalled the worker {MAX_RECOVERY_ATTEMPTS} "
+                    f"times without completing — likely a corrupt or "
+                    f"oversized input. Not re-queued. Inspect/replace the "
+                    f"file, then retry via /api/import/retry-failed."
+                ),
+            )
+            .returning(ImportTask.id)
+        )
+        poison_ids = [r[0] for r in poison_result.all()]
+
+        # Re-queue the remaining stuck rows (under the cap) and bump
+        # their recovery_count. RETURNING (id, task_type) so the requeue
+        # routes media vs archive correctly.
         stuck_result = session.execute(
             update(ImportTask)
-            .where(ImportTask.status == "processing")
-            .where(
-                ((ImportTask.task_type != "archive")
-                 & (ImportTask.started_at < media_cutoff))
-                | ((ImportTask.task_type == "archive")
-                   & (ImportTask.started_at < archive_cutoff))
-            )
+            .where(stuck_predicate)
+            .where(ImportTask.recovery_count < MAX_RECOVERY_ATTEMPTS - 1)
             .values(
                 status="queued",
                 started_at=None,
+                recovery_count=ImportTask.recovery_count + 1,
                 error="recovered from stuck state",
             )
             .returning(ImportTask.id, ImportTask.task_type)
@@ -138,7 +183,7 @@ def recover_interrupted_tasks() -> int:
             for tid, task_type in stuck:
                 enqueue_import(tid, task_type)
 
-        return len(stuck) + orphan_count
+        return len(stuck) + len(poison_ids) + orphan_count
 
 
 @celery.task(name="backend.app.tasks.maintenance.cleanup_old_tasks")
diff --git a/tests/test_maintenance.py b/tests/test_maintenance.py
index 0488943..0f5164e 100644
--- a/tests/test_maintenance.py
+++ b/tests/test_maintenance.py
@@ -164,6 +164,52 @@ def test_recover_interrupted_handles_both_stuck_and_orphans(db_sync, monkeypatch
     assert dispatched == [stuck.id]  # stuck rows re-enqueue; orphans don't
 
 
+def test_recover_interrupted_poison_pill_caps_at_max(db_sync, monkeypatch):
+    """A stuck row that's already been recovered MAX_RECOVERY_ATTEMPTS-1
+    times is marked 'failed' (with a diagnostic) instead of re-queued —
+    the circuit breaker against an input that hard-crashes the worker
+    every run. Operator-flagged 2026-05-28."""
+    from backend.app.tasks import import_file
+    from backend.app.tasks.maintenance import (
+        MAX_RECOVERY_ATTEMPTS,
+        recover_interrupted_tasks,
+    )
+    dispatched: list[int] = []
+    monkeypatch.setattr(
+        import_file.import_media_file, "delay", dispatched.append
+    )
+
+    batch_id = _make_batch(db_sync)
+    now = datetime.now(UTC)
+
+    # At the cap already (recovered MAX-1 times) → fail, don't re-queue.
+    poison = ImportTask(
+        batch_id=batch_id, source_path="/import/poison.jpg", task_type="media",
+        status="processing", started_at=now - timedelta(hours=2),
+        recovery_count=MAX_RECOVERY_ATTEMPTS - 1,
+    )
+    # One recovery short of the cap → re-queue + increment.
+    recoverable = ImportTask(
+        batch_id=batch_id, source_path="/import/ok.jpg", task_type="media",
+        status="processing", started_at=now - timedelta(hours=2),
+        recovery_count=MAX_RECOVERY_ATTEMPTS - 2,
+    )
+    db_sync.add_all([poison, recoverable])
+    db_sync.commit()
+
+    touched = recover_interrupted_tasks.apply().get()
+    assert touched == 2  # one failed + one re-queued
+
+    db_sync.refresh(poison)
+    db_sync.refresh(recoverable)
+    assert poison.status == "failed"
+    assert "corrupt or" in (poison.error or "")
+    assert recoverable.status == "queued"
+    assert recoverable.recovery_count == MAX_RECOVERY_ATTEMPTS - 1
+    # Only the recoverable row re-enqueues; the poison pill does not.
+    assert dispatched == [recoverable.id]
+
+
 def test_cleanup_old_deletes_finished_old(db_sync):
     batch_id = _make_batch(db_sync)
     now = datetime.now(UTC)

From e3cdd0f92b6592f2014d9e1b63b14bf6f0fca796 Mon Sep 17 00:00:00 2001
From: Bryan Van Deusen <bryan.vandeusen@fabledsword.com>
Date: Thu, 28 May 2026 00:01:32 -0400
Subject: [PATCH 6/7] feat(import-resilience L3): subprocess-isolated probes
 for video + archive
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Layer 3 — prevent the hard worker crash rather than just recovering from
it. The realistic process-crash vectors (operator's observed slow/heavy
tasks) are video decode and archive extraction; images decode in-process
and Pillow raises-and-skips cleanly, and a subprocess per image would
wreck deep-scan throughput, so images are intentionally not probed.

New backend/app/utils/safe_probe.py (leaf module, lazy heavy imports so
the spawned child stays light):

- probe_video(path): validates the container + first video stream via
  ffprobe (a separate binary — a decoder crash kills only ffprobe, not
  the worker). Returns width/height, which the importer didn't capture
  for videos before. crashed=True only on ffprobe timeout.
- probe_archive(path): an uncompressed-size bomb guard
  (MAX_ARCHIVE_UNCOMPRESSED_BYTES = 4 GiB) plus the format integrity
  test (zipfile.testzip / rarfile.testrar / py7zr.test) run in a
  spawned child process. A decompression-bomb OOM or native-lib
  segfault on a malformed archive shows up as a non-zero child exit
  code → crashed=True, never a dead worker.

ProbeResult.crashed distinguishes a HARD failure (subprocess killed /
timed out — the poison-pill signature → caller returns terminal
'failed') from a CLEAN rejection (corrupt-but-handled, bomb cap,
integrity mismatch → caller's choice of skipped/attached).

Wired:
- importer._import_media video branch: probe_video before the pipeline;
  crash → failed, clean reject → invalid_image skip, ok → capture dims.
- importer._import_archive: probe_archive before extract_archive; crash
  → failed, clean reject → still preserve the archive as a
  PostAttachment (matches extract_archive's fail-soft contract).
- ml.tag_and_embed video branch: probe_video before sampling 10 frames,
  so a corrupt video is rejected (status='bad_video') instead of
  crashing the ml-worker on frame decode.

Tests (test_safe_probe.py): valid/corrupt zip via probe_archive, direct
_inspect_archive size+integrity, in-process _archive_probe_target bomb
guard (monkeypatch can't reach a spawned child, so the target is called
directly), and a non-video → ok=False that's robust to ffprobe presence
in CI.
---
 backend/app/services/importer.py |  44 +++++++-
 backend/app/tasks/ml.py          |  12 +++
 backend/app/utils/safe_probe.py  | 170 +++++++++++++++++++++++++++++++
 tests/test_safe_probe.py         |  71 +++++++++++++
 4 files changed, 296 insertions(+), 1 deletion(-)
 create mode 100644 backend/app/utils/safe_probe.py
 create mode 100644 tests/test_safe_probe.py

diff --git a/backend/app/services/importer.py b/backend/app/services/importer.py
index e2b94eb..0e366d9 100644
--- a/backend/app/services/importer.py
+++ b/backend/app/services/importer.py
@@ -30,6 +30,7 @@ from ..models import (
     PostAttachment,
     Source,
 )
+from ..utils import safe_probe
 from ..utils.paths import derive_subdir, derive_top_level_artist, hash_suffixed_name
 from ..utils.phash import compute_phash, find_similar
 from ..utils.sidecar import find_sidecar, parse_sidecar
@@ -407,6 +408,29 @@ class Importer:
         return ImportResult(status="attached")
 
     def _import_archive(self, source: Path) -> ImportResult:
+        # Layer-3 isolation: bomb-size guard + integrity test in a
+        # spawned child BEFORE extracting in this process. A
+        # decompression bomb or a native-lib crash on a malformed
+        # archive is contained to the child; we reject the file cleanly
+        # instead of OOMing/segfaulting the import worker. extract_archive
+        # is already fail-soft for plain exceptions, so this only adds
+        # the hard-crash protection.
+        probe = safe_probe.probe_archive(source)
+        if not probe.ok:
+            if probe.crashed:
+                return ImportResult(
+                    status="failed",
+                    error=f"archive probe crashed/timed out: {probe.reason}",
+                )
+            # Clean rejection (bomb cap exceeded, integrity mismatch):
+            # still preserve the archive file itself as an attachment so
+            # nothing silently vanishes, matching extract_archive's
+            # fail-soft contract.
+            artist = self._resolve_artist(source)
+            post = self._post_for_sidecar(source, artist)
+            self._capture_attachment(source, post=post, artist=artist, resolved=True)
+            return ImportResult(status="attached")
+
         artist = self._resolve_artist(source)
         post = self._post_for_sidecar(source, artist)
         member_ids: list[int] = []
@@ -446,7 +470,25 @@ class Importer:
         # Compute file dimensions (images only) and apply filters.
         width = height = None
         has_alpha = False
-        if not is_video(source):
+        if is_video(source):
+            # Layer-3 isolation: validate the container via ffprobe (a
+            # separate process) before the rest of the pipeline touches
+            # it. A corrupt video that would crash a decoder is rejected
+            # cleanly here, and we capture width/height for free (the
+            # importer didn't previously record video dimensions).
+            probe = safe_probe.probe_video(source)
+            if not probe.ok:
+                if probe.crashed:
+                    return ImportResult(
+                        status="failed",
+                        error=f"video probe crashed/timed out: {probe.reason}",
+                    )
+                return ImportResult(
+                    status="skipped", skip_reason=SkipReason.invalid_image,
+                    error=probe.reason,
+                )
+            width, height = probe.width, probe.height
+        else:
             try:
                 with Image.open(source) as im:
                     im.verify()
diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py
index c4b8ddc..b114eec 100644
--- a/backend/app/tasks/ml.py
+++ b/backend/app/tasks/ml.py
@@ -71,6 +71,18 @@ def tag_and_embed(self, image_id: int) -> dict:
         embedder = get_embedder()
 
         if _is_video(src):
+            # Layer-3 isolation: ffprobe (a separate process) validates
+            # the container before we burn ~20 GPU ops sampling frames
+            # from it. A corrupt video that would crash the frame
+            # decoder is rejected cleanly here instead of taking down
+            # the ml-worker. Operator-flagged 2026-05-28.
+            from ..utils import safe_probe
+            vprobe = safe_probe.probe_video(src)
+            if not vprobe.ok:
+                return {
+                    "status": "bad_video", "image_id": image_id,
+                    "reason": vprobe.reason,
+                }
             frames = _sample_video_frames(
                 src, int(os.environ.get("VIDEO_ML_FRAMES", "10"))
             )
diff --git a/backend/app/utils/safe_probe.py b/backend/app/utils/safe_probe.py
new file mode 100644
index 0000000..02c387c
--- /dev/null
+++ b/backend/app/utils/safe_probe.py
@@ -0,0 +1,170 @@
+"""Subprocess-isolated media probes (Layer 3 of import resilience).
+
+A malformed video or archive can hard-crash the worker process — a
+decoder OOM, a native-lib segfault, or a decompression bomb. A hard
+crash leaves no terminal flip, so the recovery sweep re-queues the row
+and it crashes again: a poison-pill loop (the Layer-1 cap is the
+backstop, but isolating the crash is better — the file gets a clean
+terminal failure and the worker never dies).
+
+These probes run the risky read in a way that contains the blast:
+
+- Video: `ffprobe` is a separate binary, so a crash decoding the
+  container kills only ffprobe (non-zero exit), never the worker. Also
+  returns width/height, which the importer didn't previously capture
+  for videos.
+- Archive: an uncompressed-size guard (catches decompression bombs
+  before they OOM anything) plus an integrity test in a spawned child
+  (catches native-lib crashes on a malformed archive). A child segfault
+  / OOM shows up as a non-zero exit code, not a dead worker.
+
+Images are intentionally NOT probed here: Pillow raises (it doesn't
+segfault) on the realistic corrupt-image cases, the importer already
+catches that as an invalid_image skip, and a subprocess per image would
+wreck deep-scan throughput on a large library. Add an image branch only
+if a real image-induced worker crash is ever observed.
+
+Operator-requested 2026-05-28 (Layer 3).
+"""
+
+import json
+import multiprocessing as mp
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+
+VIDEO_PROBE_TIMEOUT_SECONDS = 60
+ARCHIVE_PROBE_TIMEOUT_SECONDS = 120
+# Refuse archives whose total UNCOMPRESSED size exceeds this — the
+# classic decompression-bomb guard (a 4 GB cap comfortably clears real
+# art-pack archives while stopping a few-KB zip that expands to TB).
+MAX_ARCHIVE_UNCOMPRESSED_BYTES = 4 * 1024 * 1024 * 1024
+
+
+@dataclass(frozen=True)
+class ProbeResult:
+    ok: bool
+    # crashed=True means the probe HARD-FAILED (subprocess killed by a
+    # signal, OOM, or timeout) — the poison-pill signature. crashed=False
+    # with ok=False means a clean rejection (corrupt-but-handled,
+    # bomb-size-exceeded, integrity mismatch). Callers map crashed → a
+    # terminal 'failed', clean → a 'skipped'/'failed' of their choosing.
+    crashed: bool = False
+    reason: str | None = None
+    width: int | None = None
+    height: int | None = None
+
+
+def probe_video(path: Path, *, timeout: float = VIDEO_PROBE_TIMEOUT_SECONDS) -> ProbeResult:
+    """Validate a video container + first video stream via ffprobe."""
+    try:
+        out = subprocess.run(
+            [
+                "ffprobe", "-v", "error",
+                "-select_streams", "v:0",
+                "-show_entries", "stream=width,height",
+                "-of", "json", str(path),
+            ],
+            capture_output=True, text=True, timeout=timeout,
+        )
+    except subprocess.TimeoutExpired:
+        return ProbeResult(ok=False, crashed=True, reason="ffprobe timed out")
+    except OSError as exc:
+        # ffprobe missing / not executable — environmental, not the
+        # file's fault. Treat as a clean non-crash failure so the import
+        # path can decide (it currently proceeds without dims).
+        return ProbeResult(ok=False, crashed=False, reason=f"ffprobe unavailable: {exc}")
+    if out.returncode != 0:
+        return ProbeResult(
+            ok=False, crashed=False,
+            reason=f"ffprobe rejected the file: {out.stderr.strip()[:200]}",
+        )
+    try:
+        streams = (json.loads(out.stdout) or {}).get("streams") or []
+    except json.JSONDecodeError as exc:
+        return ProbeResult(ok=False, crashed=False, reason=f"ffprobe output parse failed: {exc}")
+    if not streams:
+        return ProbeResult(ok=False, crashed=False, reason="no decodable video stream")
+    return ProbeResult(
+        ok=True, width=streams[0].get("width"), height=streams[0].get("height"),
+    )
+
+
+def probe_archive(path: Path, *, timeout: float = ARCHIVE_PROBE_TIMEOUT_SECONDS) -> ProbeResult:
+    """Bomb-size guard + isolated integrity test for an archive."""
+    ctx = mp.get_context("spawn")
+    q = ctx.Queue()
+    proc = ctx.Process(target=_archive_probe_target, args=(str(path), q))
+    proc.start()
+    proc.join(timeout)
+    if proc.is_alive():
+        proc.terminate()
+        proc.join(5)
+        return ProbeResult(ok=False, crashed=True, reason="archive probe timed out")
+    if proc.exitcode != 0:
+        # Negative exitcode = killed by signal (segfault); positive =
+        # the child os._exit'd or was OOM-killed. Either way the file
+        # hard-crashed the probe — the poison-pill signature.
+        return ProbeResult(
+            ok=False, crashed=True,
+            reason=f"archive probe crashed (exit {proc.exitcode})",
+        )
+    try:
+        outcome = q.get(timeout=5)
+    except Exception:  # noqa: BLE001 — empty queue / broken pipe
+        return ProbeResult(ok=False, crashed=True, reason="archive probe produced no result")
+    status, detail = outcome
+    if status == "ok":
+        return ProbeResult(ok=True)
+    return ProbeResult(ok=False, crashed=False, reason=detail)
+
+
+def _archive_probe_target(path_str: str, q) -> None:
+    """Runs in the spawned child. Reads member sizes (bomb guard) then
+    runs the format's integrity test. Puts ('ok', None) or
+    ('error', reason). A crash/OOM here never reaches the queue — the
+    parent reads the non-zero exit code instead."""
+    path = Path(path_str)
+    ext = path.suffix.lower()
+    try:
+        total, test_bad = _inspect_archive(path, ext)
+    except Exception as exc:  # noqa: BLE001 — clean rejection
+        q.put(("error", f"{type(exc).__name__}: {exc}"))
+        return
+    if total is not None and total > MAX_ARCHIVE_UNCOMPRESSED_BYTES:
+        gib = total / (1024 ** 3)
+        q.put(("error", f"uncompressed size {gib:.1f} GiB exceeds the bomb-guard cap"))
+        return
+    if test_bad is not None:
+        q.put(("error", f"integrity test failed at member {test_bad!r}"))
+        return
+    q.put(("ok", None))
+
+
+def _inspect_archive(path: Path, ext: str):
+    """Return (total_uncompressed_bytes | None, first_bad_member | None)
+    for the archive. Format-specific; raises on a structurally-broken
+    container (caught by the child as a clean rejection)."""
+    if ext in (".zip", ".cbz"):
+        import zipfile
+
+        with zipfile.ZipFile(path) as zf:
+            total = sum(zi.file_size for zi in zf.infolist())
+            return total, zf.testzip()
+    if ext == ".rar":
+        import rarfile
+
+        with rarfile.RarFile(path) as rf:
+            total = sum(getattr(ri, "file_size", 0) for ri in rf.infolist())
+            rf.testrar()
+            return total, None
+    if ext == ".7z":
+        import py7zr
+
+        with py7zr.SevenZipFile(path, "r") as zf:
+            info = zf.archiveinfo()
+            total = getattr(info, "uncompressed", None)
+            ok = zf.test()  # True / None when all members pass
+            return total, (None if ok in (True, None) else "7z test reported corruption")
+    # Unknown extension — nothing to test; treat as clean.
+    return None, None
diff --git a/tests/test_safe_probe.py b/tests/test_safe_probe.py
new file mode 100644
index 0000000..6dd7ea2
--- /dev/null
+++ b/tests/test_safe_probe.py
@@ -0,0 +1,71 @@
+"""Layer-3 subprocess-isolated probe tests.
+
+The bomb-guard cap is exercised against `_archive_probe_target` directly
+(in-process, where a monkeypatch on the module constant takes effect) —
+spawn re-imports the module in the child, so a parent-process
+monkeypatch wouldn't reach the spawned worker.
+"""
+
+import multiprocessing as mp
+import zipfile
+from pathlib import Path
+
+from backend.app.utils import safe_probe
+
+
+def _zip(path, entries):
+    with zipfile.ZipFile(path, "w") as zf:
+        for name, data in entries.items():
+            zf.writestr(name, data)
+
+
+def test_probe_archive_valid_zip(tmp_path):
+    z = tmp_path / "ok.zip"
+    _zip(z, {"a.jpg": b"hello", "b.png": b"world"})
+    res = safe_probe.probe_archive(z)
+    assert res.ok is True
+    assert res.crashed is False
+
+
+def test_probe_archive_corrupt_zip_clean_rejection(tmp_path):
+    z = tmp_path / "broken.zip"
+    z.write_bytes(b"PK\x03\x04 not really a zip past here")
+    res = safe_probe.probe_archive(z)
+    assert res.ok is False
+    # Corrupt-but-handled (zipfile raises BadZipFile in the child) — a
+    # clean rejection, not a hard crash.
+    assert res.crashed is False
+    assert res.reason
+
+
+def test_inspect_archive_reports_size_and_clean_integrity(tmp_path):
+    z = tmp_path / "sized.zip"
+    _zip(z, {"a.txt": b"x" * 100, "b.txt": b"y" * 50})
+    total, bad = safe_probe._inspect_archive(z, ".zip")
+    assert total == 150
+    assert bad is None
+
+
+def test_archive_probe_target_bomb_guard(tmp_path, monkeypatch):
+    """In-process call to the child target so the monkeypatched cap
+    takes effect. A normal zip whose uncompressed size exceeds the
+    (lowered) cap is rejected with the bomb-guard reason."""
+    monkeypatch.setattr(safe_probe, "MAX_ARCHIVE_UNCOMPRESSED_BYTES", 10)
+    z = tmp_path / "bomb.zip"
+    _zip(z, {"big.txt": b"x" * 5000})  # 5000 uncompressed > 10-byte cap
+    q = mp.get_context("spawn").Queue()
+    safe_probe._archive_probe_target(str(z), q)
+    status, detail = q.get(timeout=5)
+    assert status == "error"
+    assert "bomb-guard cap" in detail
+
+
+def test_probe_video_non_video_is_not_ok(tmp_path):
+    """A text file is not a decodable video. Whether ffprobe is present
+    (returncode != 0) or absent (OSError → 'unavailable'), the result is
+    ok=False. We don't assert on crashed/reason so the test is robust to
+    ffprobe presence in CI."""
+    f = tmp_path / "nope.txt"
+    f.write_text("definitely not a video container")
+    res = safe_probe.probe_video(f)
+    assert res.ok is False

From dcfe55d731ad0e1b25037e1d704f60a6bf179dd4 Mon Sep 17 00:00:00 2001
From: Bryan Van Deusen <bryan.vandeusen@fabledsword.com>
Date: Thu, 28 May 2026 00:08:03 -0400
Subject: [PATCH 7/7] feat(import-resilience L2): one-shot re-download for
 corrupt downloaded files
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Layer 2 — remediate a corrupt file by re-fetching a fresh copy from its
source, bounded to a single attempt. Operator-requested 2026-05-28.

New backend/app/services/refetch_service.py:
- resolve_refetch_source: parse the failed file's sidecar → platform,
  derive the artist from the import path, find an ENABLED Source with a
  real feed URL for (artist, platform). Returns None for filesystem-only
  imports, missing sidecars, or `sidecar:<platform>:<slug>` synthetic
  anchors (not pollable).
- attempt_refetch: if not already refetched AND a Source resolves,
  delete the corrupt file (so gallery-dl's skip_existing re-fetches it),
  set ImportTask.refetched=True, and trigger ONE download_source
  re-check. Bounded by `refetched` so source-side corruption can't loop.

Wiring:
- Manual endpoint POST /api/import/tasks/<id>/refetch (only on 'failed'
  tasks). Returns refetch_queued / no_source / already_refetched /
  not_found / not_failed.
- Auto path in recover_interrupted_tasks: for each poison-pill row, if
  env FC_AUTO_REFETCH_CORRUPT=1, attempt_refetch (default OFF — the
  manual button is the primary path; auto is opt-in since re-fetch
  deletes a file + re-runs the downloader).
- Frontend: a cloud-refresh icon button on failed rows in ImportTaskList
  → stores.import.refetchTask → toast keyed on the result status.

Filesystem imports with no upstream return no_source — the operator's
only remediation there is replacing the file on disk, surfaced clearly
in the toast.

Tests: 404 unknown task, 400 non-failed task, no_source when
unresolvable, and the full resolvable-source path (file deleted,
refetched flag set, one download_source dispatched, second call is a
no-op). The resolvable test repoints the migration-seeded
import_settings(id=1) scan path rather than inserting a conflicting row.
---
 backend/app/api/import_admin.py               |  37 ++++++
 backend/app/services/refetch_service.py       | 106 +++++++++++++++++
 backend/app/tasks/maintenance.py              |  23 ++++
 .../components/settings/ImportTaskList.vue    |  35 +++++-
 frontend/src/stores/import.js                 |  12 +-
 tests/test_api_import_admin.py                | 109 ++++++++++++++++++
 6 files changed, 320 insertions(+), 2 deletions(-)
 create mode 100644 backend/app/services/refetch_service.py

diff --git a/backend/app/api/import_admin.py b/backend/app/api/import_admin.py
index 2ebeb9b..7ad3fa7 100644
--- a/backend/app/api/import_admin.py
+++ b/backend/app/api/import_admin.py
@@ -128,6 +128,43 @@ async def retry_failed():
     return jsonify({"retried": len(failed)})
 
 
+@import_admin_bp.route("/tasks/<int:task_id>/refetch", methods=["POST"])
+async def refetch_task(task_id: int):
+    """Layer-2 one-shot re-download: delete the (corrupt) file behind a
+    failed import task and re-run its source's downloader to fetch a
+    fresh copy. Only works for files that resolve to an enabled,
+    real-URL subscription Source; filesystem-only imports return
+    no_source.
+
+    Returns one of: refetch_queued (+source_id) / no_source /
+    already_refetched / not_found / not_failed.
+    """
+    async with get_session() as session:
+        result = await session.run_sync(_refetch_task_sync, task_id)
+    if result["status"] == "not_found":
+        return jsonify(result), 404
+    if result["status"] == "not_failed":
+        return jsonify(result), 400
+    return jsonify(result)
+
+
+def _refetch_task_sync(session, task_id: int) -> dict:
+    from pathlib import Path
+
+    from ..models import ImportSettings
+    from ..services.refetch_service import attempt_refetch
+
+    task = session.get(ImportTask, task_id)
+    if task is None:
+        return {"status": "not_found"}
+    if task.status != "failed":
+        return {"status": "not_failed"}
+    settings = session.execute(
+        select(ImportSettings).where(ImportSettings.id == 1)
+    ).scalar_one()
+    return attempt_refetch(session, task, Path(settings.import_scan_path))
+
+
 @import_admin_bp.route("/clear-stuck", methods=["POST"])
 async def clear_stuck():
     """Force any non-terminal ImportTask (status in pending/queued/
diff --git a/backend/app/services/refetch_service.py b/backend/app/services/refetch_service.py
new file mode 100644
index 0000000..41832bc
--- /dev/null
+++ b/backend/app/services/refetch_service.py
@@ -0,0 +1,106 @@
+"""Layer-2 one-shot re-download remediation for corrupt imported files.
+
+When an import fails on a file that came from a known, pollable
+subscription Source, deleting the bad copy and re-running the source's
+downloader can fetch a fresh, unblemished copy. This only helps when:
+
+  - the corruption is in transit / on disk (not at the source), AND
+  - the file resolves to an ENABLED Source with a real feed URL
+    (a `sidecar:<platform>:<slug>` synthetic anchor is not pollable),
+    AND
+  - we haven't already re-fetched this task once (bounded by
+    ImportTask.refetched so source-side corruption can't loop).
+
+Filesystem-only imports with no resolvable Source return 'no_source' —
+the operator's only remediation there is to replace the file on disk.
+
+Operator-requested 2026-05-28 (Layer 2).
+"""
+
+import json
+import logging
+from pathlib import Path
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from ..models import Artist, ImportTask, Source
+from ..utils.paths import derive_top_level_artist
+from ..utils.sidecar import find_sidecar, parse_sidecar
+from ..utils.slug import slugify
+
+log = logging.getLogger(__name__)
+
+
+def resolve_refetch_source(
+    session: Session, source_path: str, import_root: Path,
+) -> Source | None:
+    """Find an enabled, real-URL Source for the file's (artist, platform),
+    or None when nothing re-pollable resolves."""
+    path = Path(source_path)
+    sc = find_sidecar(path)
+    if sc is None:
+        return None
+    try:
+        data = json.loads(sc.read_text("utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return None
+    if not isinstance(data, dict):
+        return None
+    sd = parse_sidecar(data)
+    if not sd.platform:
+        return None
+    artist_name = derive_top_level_artist(path, import_root)
+    if not artist_name:
+        return None
+    artist = session.execute(
+        select(Artist).where(Artist.slug == slugify(artist_name))
+    ).scalar_one_or_none()
+    if artist is None:
+        return None
+    src = session.execute(
+        select(Source)
+        .where(
+            Source.artist_id == artist.id,
+            Source.platform == sd.platform,
+            Source.enabled.is_(True),
+        )
+        .order_by(Source.id.asc())
+    ).scalars().first()
+    if src is None:
+        return None
+    if (src.url or "").startswith("sidecar:"):
+        return None  # synthetic anchor — not a pollable feed
+    return src
+
+
+def attempt_refetch(
+    session: Session, task: ImportTask, import_root: Path,
+) -> dict:
+    """Delete the corrupt file, mark the task refetched, and trigger ONE
+    source re-check. Idempotent/bounded: a task already refetched (or
+    with no resolvable Source) is a no-op. Commits."""
+    if task.refetched:
+        return {"status": "already_refetched"}
+    src = resolve_refetch_source(session, task.source_path, import_root)
+    if src is None:
+        return {"status": "no_source"}
+
+    # Remove the bad copy so gallery-dl (skip_existing) re-fetches it on
+    # the source re-check instead of skipping the still-present corrupt
+    # file.
+    try:
+        Path(task.source_path).unlink(missing_ok=True)
+    except OSError as exc:
+        log.warning("refetch unlink failed for %s: %s", task.source_path, exc)
+
+    task.refetched = True
+    session.add(task)
+    session.commit()
+
+    # Lazy import to avoid a tasks→services→tasks import cycle at module
+    # load. download_source.delay() is sync-safe in any context.
+    from ..tasks.download import download_source
+
+    download_source.delay(src.id)
+    return {"status": "refetch_queued", "source_id": src.id}
diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py
index f864873..4584d01 100644
--- a/backend/app/tasks/maintenance.py
+++ b/backend/app/tasks/maintenance.py
@@ -1,6 +1,7 @@
 """Periodic maintenance: recover stuck import tasks, garbage-collect old finished tasks."""
 
 import logging
+import os
 import subprocess
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
@@ -183,6 +184,28 @@ def recover_interrupted_tasks() -> int:
             for tid, task_type in stuck:
                 enqueue_import(tid, task_type)
 
+        # Layer-2 auto re-download (env-gated, default OFF). For each
+        # poison-pill row that resolves to a pollable Source, delete the
+        # bad file and trigger ONE source re-check to fetch a fresh
+        # copy. Bounded by ImportTask.refetched so source-side
+        # corruption can't loop. The 'failed' row stays as history; the
+        # re-downloaded file re-imports as a fresh task on the next scan.
+        if poison_ids and os.environ.get("FC_AUTO_REFETCH_CORRUPT", "0") == "1":
+            from ..models import ImportSettings
+            from ..services.refetch_service import attempt_refetch
+            import_root = Path(session.execute(
+                select(ImportSettings.import_scan_path)
+                .where(ImportSettings.id == 1)
+            ).scalar_one())
+            for pid in poison_ids:
+                ptask = session.get(ImportTask, pid)
+                if ptask is None:
+                    continue
+                try:
+                    attempt_refetch(session, ptask, import_root)
+                except Exception as exc:  # noqa: BLE001 — best-effort
+                    log.warning("auto-refetch failed for task %s: %s", pid, exc)
+
         return len(stuck) + len(poison_ids) + orphan_count
 
 
diff --git a/frontend/src/components/settings/ImportTaskList.vue b/frontend/src/components/settings/ImportTaskList.vue
index 97bbebf..4eadf3c 100644
--- a/frontend/src/components/settings/ImportTaskList.vue
+++ b/frontend/src/components/settings/ImportTaskList.vue
@@ -51,6 +51,19 @@
           title="Click for full error"
         >{{ shorten(item.error, 60) }}</button>
       </template>
+      <template #item.actions="{ item }">
+        <v-btn
+          v-if="item.status === 'failed'"
+          icon size="x-small" variant="text"
+          :loading="refetching === item.id"
+          @click="onRefetch(item)"
+        >
+          <v-icon size="small">mdi-cloud-refresh</v-icon>
+          <v-tooltip activator="parent" location="top">
+            Re-fetch original (re-download from source)
+          </v-tooltip>
+        </v-btn>
+      </template>
     </v-data-table-virtual>
     <div v-if="store.hasMore" class="d-flex justify-center py-3">
       <v-btn variant="text" size="small" @click="onLoadMore">Load more</v-btn>
@@ -149,9 +162,29 @@ const headers = [
   { title: 'Source', key: 'source_path', sortable: false },
   { title: 'Size', key: 'size_bytes', sortable: false, width: 90 },
   { title: 'Created', key: 'created_at', sortable: false, width: 150 },
-  { title: 'Note', key: 'error', sortable: false }
+  { title: 'Note', key: 'error', sortable: false },
+  { title: '', key: 'actions', sortable: false, width: 56 }
 ]
 
+const refetching = ref(null)
+const _REFETCH_MSG = {
+  refetch_queued: { text: 'Re-fetch queued — re-downloading from source', type: 'success' },
+  no_source: { text: 'No re-fetchable source (filesystem import — replace the file manually)', type: 'info' },
+  already_refetched: { text: 'Already re-fetched once', type: 'info' },
+}
+async function onRefetch(item) {
+  refetching.value = item.id
+  try {
+    const res = await store.refetchTask(item.id)
+    const msg = _REFETCH_MSG[res.status] || { text: `Re-fetch: ${res.status}`, type: 'info' }
+    window.__fcToast?.(msg)
+  } catch (e) {
+    window.__fcToast?.({ text: `Re-fetch failed: ${e.message}`, type: 'error' })
+  } finally {
+    refetching.value = null
+  }
+}
+
 const hasFailed = computed(() => store.tasks.some(t => t.status === 'failed'))
 const hasStuck = computed(() => store.tasks.some(
   t => t.status === 'pending' || t.status === 'queued' || t.status === 'processing'
diff --git a/frontend/src/stores/import.js b/frontend/src/stores/import.js
index 64e5146..98d4d56 100644
--- a/frontend/src/stores/import.js
+++ b/frontend/src/stores/import.js
@@ -146,6 +146,15 @@ export const useImportStore = defineStore('import', () => {
     return body
   }
 
+  // Layer-2 one-shot re-download for a failed task's (corrupt) file.
+  // Returns the endpoint's status dict (refetch_queued / no_source /
+  // already_refetched). Caller surfaces it as a toast.
+  async function refetchTask(taskId) {
+    const body = await api.post(`/api/import/tasks/${taskId}/refetch`)
+    await loadTasks(true)
+    return body
+  }
+
   const hasMore = computed(() => tasksNextCursor.value !== null)
 
   return {
@@ -155,6 +164,7 @@ export const useImportStore = defineStore('import', () => {
     triggerError,
     loadSettings, patchSettings,
     refreshStatus, triggerScan,
-    loadTasks, setStatusFilter, retryFailed, clearCompleted, clearStuck
+    loadTasks, setStatusFilter, retryFailed, clearCompleted, clearStuck,
+    refetchTask,
   }
 })
diff --git a/tests/test_api_import_admin.py b/tests/test_api_import_admin.py
index f5512aa..f033fd5 100644
--- a/tests/test_api_import_admin.py
+++ b/tests/test_api_import_admin.py
@@ -171,6 +171,115 @@ async def test_trigger_still_rejects_unknown_mode(client):
     assert resp.status_code == 400
 
 
+@pytest.mark.asyncio
+async def test_refetch_404_for_unknown_task(client):
+    resp = await client.post("/api/import/tasks/999999/refetch")
+    assert resp.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_refetch_400_for_non_failed_task(client, db):
+    batch = ImportBatch(triggered_by="manual", source_path="/import", scan_mode="quick")
+    db.add(batch)
+    await db.flush()
+    task = ImportTask(
+        batch_id=batch.id, source_path="/x.jpg", task_type="media",
+        status="complete", finished_at=datetime.now(UTC),
+    )
+    db.add(task)
+    await db.commit()
+    resp = await client.post(f"/api/import/tasks/{task.id}/refetch")
+    assert resp.status_code == 400
+    assert (await resp.get_json())["status"] == "not_failed"
+
+
+@pytest.mark.asyncio
+async def test_refetch_no_source_when_unresolvable(client, db):
+    """A failed task whose file has no sidecar / no resolvable Source
+    returns no_source (filesystem-only import — nothing to re-poll)."""
+    batch = ImportBatch(triggered_by="manual", source_path="/import", scan_mode="quick")
+    db.add(batch)
+    await db.flush()
+    task = ImportTask(
+        batch_id=batch.id, source_path="/import/nowhere/x.jpg",
+        task_type="media", status="failed", finished_at=datetime.now(UTC),
+    )
+    db.add(task)
+    await db.commit()
+    resp = await client.post(f"/api/import/tasks/{task.id}/refetch")
+    assert resp.status_code == 200
+    assert (await resp.get_json())["status"] == "no_source"
+
+
+@pytest.mark.asyncio
+async def test_refetch_queued_with_resolvable_source(client, db, tmp_path, monkeypatch):
+    """A failed task whose file resolves (via sidecar → artist+platform)
+    to an enabled, real-URL Source: the file is deleted, the task is
+    marked refetched, and ONE source re-check is queued."""
+    import json as _json
+
+    from sqlalchemy import update as _update
+
+    from backend.app.models import Artist, ImportSettings, Source
+    from backend.app.tasks import download as download_mod
+
+    # Stub the downloader so the eager test doesn't run a real fetch.
+    dispatched = []
+    monkeypatch.setattr(download_mod.download_source, "delay", dispatched.append)
+
+    # import_root/<ArtistName>/post.jpg + sidecar identifying the platform.
+    import_root = tmp_path / "import"
+    artist_dir = import_root / "Maewix"
+    artist_dir.mkdir(parents=True)
+    media = artist_dir / "post.jpg"
+    media.write_bytes(b"corrupt-bytes")
+    (artist_dir / "post.jpg.json").write_text(
+        _json.dumps({"category": "patreon", "post_id": 123})
+    )
+
+    # import_settings(id=1) is migration-seeded; point its scan path at
+    # our tmp import root rather than inserting a conflicting row.
+    await db.execute(
+        _update(ImportSettings).where(ImportSettings.id == 1)
+        .values(import_scan_path=str(import_root))
+    )
+    artist = Artist(name="Maewix", slug="maewix")
+    db.add(artist)
+    await db.flush()
+    db.add(Source(
+        artist_id=artist.id, platform="patreon",
+        url="https://www.patreon.com/maewix", enabled=True,
+        config_overrides={},
+    ))
+    batch = ImportBatch(triggered_by="manual", source_path=str(import_root), scan_mode="quick")
+    db.add(batch)
+    await db.flush()
+    task = ImportTask(
+        batch_id=batch.id, source_path=str(media), task_type="media",
+        status="failed", finished_at=datetime.now(UTC),
+    )
+    db.add(task)
+    await db.commit()
+
+    resp = await client.post(f"/api/import/tasks/{task.id}/refetch")
+    assert resp.status_code == 200
+    body = await resp.get_json()
+    assert body["status"] == "refetch_queued"
+    assert len(dispatched) == 1
+    assert not media.exists()  # corrupt copy removed for re-fetch
+
+    from sqlalchemy import select as _select
+    refetched = (await db.execute(
+        _select(ImportTask.refetched).where(ImportTask.id == task.id)
+    )).scalar_one()
+    assert refetched is True
+
+    # Second attempt is a no-op (bounded to one).
+    resp2 = await client.post(f"/api/import/tasks/{task.id}/refetch")
+    assert (await resp2.get_json())["status"] == "already_refetched"
+    assert len(dispatched) == 1
+
+
 @pytest.mark.asyncio
 async def test_trigger_accepts_verify(client, monkeypatch):
     # Stub the verify task's dispatch so the API contract is asserted