Integrity remediation loop: verify before swap

README.md

@@ -38,7 +38,15 @@ Episode Guard can inspect upcoming episodes already on disk during lookahead che
 ### How it works:
 1. **Tier 1 (Runtime Validation):** Compares physical file duration against expected duration from Sonarr metadata. Flagged if actual duration is below the tolerance threshold (default: `< 80%`).
 2. **Tier 2 (Stream & Tail Probe):** If media storage is mounted, runs `ffprobe` and decodes the final 30 seconds of the video stream via `ffmpeg` to `/dev/null` (`-f null -`) to catch packet corruption and broken EOF.
-3. **Auto-Remediation:** If a bad file is detected and `integrity_check_mode` is set to `Auto-Remediate`, Episode Guard deletes the bad `episodeFile` in Sonarr via API and immediately triggers a replacement search. If set to `Notify Only`, it sends an alert and records the finding without deleting.
+3. **Auto-Remediation (verify before swap):** A flagged file first gets a full-stream
+   decode to confirm it is genuinely corrupt. If confirmed, Episode Guard keeps the
+   current file and asks Sonarr for a replacement. The replacement is full-decoded in
+   the downloads folder before anything is touched; only then is the old file swapped
+   out via Sonarr manual import, and the final file is decoded once more. Failed
+   replacements are blocklisted, up to 3 different releases, then a summary alert.
+   Falls back to the old delete-first behaviour (with a warning) if the downloads
+   folder is not mounted. Exhausted remediations can be cleared via
+   `POST /api/remediation/:episodeId/reset`.
 4. **Verification Cache:** Clean files are cached in SQLite so files are only probed once.
 
 To enable local stream probing, bind-mount your TV media folder read-only in `docker-compose.yml`:
@@ -47,8 +55,11 @@ To enable local stream probing, bind-mount your TV media folder read-only in `do
 volumes:
   - ./data:/data
   - /data/TV:/tv:ro
+  - /data/Downloads:/downloads:ro   # downloads root incl. unpackerr output
 ```
 
+Verify-before-swap needs this mount plus `DOWNLOADS_DIR=/downloads`. Without it, remediation falls back to delete-first.
+
 ## Activity log
 
 | Event | What it means |

docs/superpowers/specs/2026-09-20-integrity-remediation-loop-design.md

@@ -0,0 +1,154 @@
+# Integrity remediation loop: verify before swap
+
+Date: 2026-09-20
+Status: approved design, pending implementation plan
+
+## Problem
+
+The current auto-remediation deletes a corrupt `episodeFile` in Sonarr, then triggers a replacement search. If the search is slow, produces another corrupt download, or fails outright, the episode is simply missing. In practice the media server ended up with no file and no warning until the next integrity pass.
+
+Real-world trigger: the daily log sweep flagged 45 integrity errors across ~15 distinct episodes (It's Always Sunny S11-S13, Chicago P.D.). Some of those flags come from the Tier 2 tail decode (the `-ss` decode of the final 30 seconds), which can report a failure on files that decode fine end to end. Today that ambiguity can lead to deleting a file that was never actually broken.
+
+## Goal
+
+Replace delete-first remediation with a verify-before-swap loop:
+
+1. Confirm a flagged file is genuinely corrupt with a full decode.
+2. Get a replacement downloaded, still keeping the old file on disk.
+3. Full-decode the replacement before it replaces anything.
+4. Only swap after the replacement passes.
+5. Give up cleanly after 3 different releases and say so.
+
+No moment in this loop should leave the episode with no file on disk.
+
+## Scope
+
+Watch-ahead (upcoming) episodes only, matching the existing integrity-check scope. Library-wide auditing is out of scope for this feature.
+
+## Current state (what exists today)
+
+- Tier 1: physical duration vs Sonarr metadata, tolerance `< 80%` of expected.
+- Tier 2: `ffprobe` plus a 30-second tail decode to `/dev/null`, run only when media storage is mounted.
+- `integrity_check_mode`: `Auto-Remediate` (delete + re-search now) or `Notify Only`.
+- Verification cache in SQLite so clean files are not re-probed.
+- Sonarr API client, Apprise notifications, activity feed UI.
+
+## Design
+
+### Pipeline
+
+```
+flagged by Tier 1 or Tier 2
+  -> CONFIRM: full decode of the whole file (ffmpeg -f null -)
+       pass  -> clear the flag, cache result, done
+       fail  -> enter remediation state
+  -> SEARCH: trigger a Sonarr episode search, keep the old file
+  -> VERIFY: poll the Sonarr queue; when the download for this episode
+             reaches a completed state, full-decode the queued file
+       clean -> SWAP: force-import via Sonarr manual import (Sonarr's
+                      media management moves it to the final folder,
+                      replacing the old file in one step)
+       corrupt -> remove the queue item with blocklist=true, retry
+  -> POST-IMPORT: full-decode the file now in the final /tv folder
+       clean -> swapped, cache the result
+       corrupt -> notify (move should be byte-identical; treat as anomaly)
+  -> after 3 blocked releases: EXHAUSTED, notify-only summary
+```
+
+### Full decode
+
+One function, used for both the confirm pass and the replacement verify:
+
+- `ffmpeg -i <file> -f null -` over the entire file, audio and video streams, error log captured.
+- Timeout scaled to file length (default: expected duration + 5 minutes, capped).
+- Result recorded in the verification cache keyed by file path + size + mtime, so a queued download is never decoded twice.
+- Extraction stability: with unpackerr in the pipeline, a queue item can flip to "completed" while the extraction is still writing. Before decoding, require the file's size to be unchanged across two polls spaced a few seconds apart; until then keep polling.
+
+### Delete-after guarantee
+
+Sonarr's auto-import refuses a same-quality download while the episode already has a file ("not an upgrade"), so the loop never relies on it. The swap uses Sonarr's manual import API (`DownloadedEpisodesScan` on the verified output path, or `manualimport` with the episode ID) — Sonarr's media management then moves the file into the final folder and replaces the old file as one operation.
+
+Fallback: if Sonarr still refuses the same-quality replacement on manual import, the loop deletes the old `episodeFile` first and imports the verified queued file immediately after. That leaves a seconds-long gap, but only ever after the replacement has passed the full decode, so no verified-good file is at risk.
+
+The old file is never removed before a verified clean replacement exists. Enforced in one place (the swap step); every other path keeps existing behaviour.
+
+### Post-import verification
+
+After Sonarr's media management move, the final file in `/tv` is full-decoded again before the episode is marked `swapped`. The move should be byte-identical (copy/hardlink), so this is belt-and-braces against anything odd in the pipeline, not a corruption filter. A failure here is treated as an anomaly: notify-only, do not loop.
+
+### Release dedup
+
+Each failed attempt adds a Sonarr blocklist entry (`removeFromClient: true, blocklist: true` on queue deletion). Attempt N+1 therefore picks a different release. Attempt counter and failed release IDs stored per episode in the remediation state.
+
+### State machine (SQLite, per episode)
+
+`suspect -> confirmed -> searching -> verifying -> importing -> swapped | exhausted`
+
+- Rows survive container restarts; the watcher resumes polling on boot.
+- Any queue item stuck in a non-completed state longer than `REMEDIATION_QUEUE_TIMEOUT_HOURS` (default: 6) counts as a failed attempt: remove with blocklist, retry.
+- If Sonarr reports the download failed before completion, same handling.
+
+### Fallback
+
+If the completed-downloads directory is not mounted (or the queue item's `outputPath` is unreadable), the loop cannot verify the replacement. In that case it falls back to today's delete-first behaviour and says so explicitly in the Apprise notification and the activity row. Silent degradation is not acceptable.
+
+### Configuration
+
+New env vars, all with sane defaults so the feature works without touching compose:
+
+| Var | Default | Purpose |
+|---|---|---|
+| `DOWNLOADS_DIR` | unset | Read-only mount of the downloads root (the extracted location Sonarr imports from, including anything unpackerr writes); presence enables queue-file verification |
+| `REMEDIATION_MAX_ATTEMPTS` | `3` | Distinct releases tried before exhausted |
+| `REMEDIATION_QUEUE_TIMEOUT_HOURS` | `6` | Stuck queue item treated as failed |
+
+Existing `integrity_check_mode` keeps its two values; the verify-before-swap loop replaces what `Auto-Remediate` does internally. No new mode names to configure.
+
+### Mounting (deploy side)
+
+EpisodeGuard already mounts `/tv:ro`. Deployment adds one line so queued downloads can be probed:
+
+```yaml
+volumes:
+  - ./data:/data
+  - /data/TV:/tv:ro
+  - /data/Downloads:/downloads:ro   # downloads root incl. unpackerr output — NOT the /tv destination
+```
+
+`DOWNLOADS_DIR=/downloads` then points the verifier at the right place: the probe target is always the Sonarr queue item's `outputPath`, which with unpackerr is the extracted folder inside the downloads root — never the final `/tv` destination (the swap has not happened yet) and never the raw archive (Sonarr only marks the item completed after unpackerr finishes). If the queue item's absolute path starts with a different mount point, the code maps the `outputPath` prefix onto this mount rather than requiring identical paths.
+
+### UI and notifications
+
+- Activity rows show the loop state (`confirming`, `verifying`, `swapped`, `exhausted`) and per-attempt detail in the plain-English details column.
+- Apprise messages: one on confirmed-corrupt (with full-decode result), one on successful swap, one on exhaustion listing all 3 blocked releases.
+- No new UI pages; the existing activity feed carries it.
+
+### Interactions and error handling
+
+- Sonarr unreachable during the loop: poll retries with backoff, state unchanged, logged.
+- The swap step deletes the old file via the existing Sonarr API path; if that call fails after the replacement is verified, log the error and retry the delete on the next poll cycle (the verified queued file stays queued; Sonarr still holds it).
+- Old file already gone when the swap step runs (user deleted it manually): proceed with import only.
+- Verification cache hits never skip a remediation decision; cache only short-circuits re-decoding the same unchanged file.
+
+### Testing
+
+- Unit: state transitions, attempt counting, blocklist calls, timeout handling, path-prefix mapping.
+- Integration (mock Sonarr + a real ffmpeg run on a sample broken MP4): confirm pass rejects the broken file, accepts a healthy one, swap only happens after verify passes.
+- Manual: run against the live Sunny/Chicago P.D. backlog once implemented, watch one full loop end to end.
+
+### Loop prevention
+
+Four invariants guarantee every path through the flow terminates:
+
+1. **Attempt cap with blocklist**: max 3 attempts, each blocked release can never be grabbed again, so the attempt sequence is strictly decreasing in candidates.
+2. **Terminal states are sticky**: `exhausted` never auto-retries; clearing it requires a manual reset from the UI. `swapped` is terminal — later lookahead scans may re-flag the file, but the confirm pass full-decodes it, it passes (it was verified twice), and nothing re-enters the loop.
+3. **Every waiting state has a time bound**: `verifying` queue timeout (default 6h) and a `searching` no-results timeout (default 24h). An episode with no eligible releases anywhere ends up `exhausted`, not stuck in `searching` forever.
+4. **Post-import failure never loops**: a corrupt file after the media-management move is notify-only, full stop. And a global concurrency cap of 1 concurrent remediation keeps a batch of corrupt episodes from spawning parallel loops hammering Sonarr and ffmpeg; the rest wait in line.
+
+False-corrupt guard: the confirm pass retries a decode once on timeout or IO-class errors before declaring a file corrupt, so a flaky disk or a sleeping drive cannot burn attempts on a healthy file.
+
+## Out of scope
+
+- Library-wide audits.
+- Movies (Sonarr-only today).
+- Re-encoding or repairing corrupt files.

src/config.js

@@ -9,6 +9,10 @@ export const env = {
   secretKey:      process.env.SECRET_KEY,
   appUrl:         (process.env.APP_URL ?? 'http://localhost:8988').replace(/\/$/, ''),
   dataDir:        process.env.DATA_DIR ?? '/data',
+  downloadsDir:   process.env.DOWNLOADS_DIR || '',
+  remediationMaxAttempts: parseInt(process.env.REMEDIATION_MAX_ATTEMPTS ?? '3', 10),
+  remediationQueueTimeoutHours: parseInt(process.env.REMEDIATION_QUEUE_TIMEOUT_HOURS ?? '6', 10),
+  remediationSearchingTimeoutHours: parseInt(process.env.REMEDIATION_SEARCHING_TIMEOUT_HOURS ?? '24', 10),
   port:           parseInt(process.env.PORT ?? '8988', 10),
 };
 

src/db.js

@@ -67,6 +67,22 @@ function initSchema(db) {
       checked_at    TEXT    NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
     );
 
+    CREATE TABLE IF NOT EXISTS remediation_episodes (
+      episode_id    INTEGER PRIMARY KEY,
+      series_id     INTEGER NOT NULL,
+      show_title    TEXT    NOT NULL,
+      season        INTEGER NOT NULL,
+      episode       INTEGER NOT NULL,
+      old_file_id   INTEGER,
+      reason        TEXT,
+      state         TEXT    NOT NULL DEFAULT 'queued'
+                    CHECK (state IN ('queued','searching','verifying','importing','swapped','exhausted')),
+      attempts      INTEGER NOT NULL DEFAULT 0,
+      blocked_releases TEXT NOT NULL DEFAULT '[]',
+      details       TEXT,
+      state_changed_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ','now'))
+    );
+
     CREATE INDEX IF NOT EXISTS idx_activity_ts   ON activity_log(timestamp);
     CREATE INDEX IF NOT EXISTS idx_activity_type ON activity_log(event_type);
     CREATE INDEX IF NOT EXISTS idx_activity_show ON activity_log(show_title);
@@ -294,3 +310,48 @@ export function clearVerifiedFile(fileId) {
   if (!fileId) return;
   getDb().prepare('DELETE FROM verified_files WHERE file_id = ?').run(fileId);
 }
+
+// Remediation state machine persistence
+
+export function createRemediation({ seriesId, episodeId, showTitle, season, episode, oldFileId, reason }) {
+  const db = getDb();
+  db.prepare(`INSERT INTO remediation_episodes
+    (episode_id, series_id, show_title, season, episode, old_file_id, reason, state)
+    VALUES (?, ?, ?, ?, ?, ?, ?, 'queued')
+    ON CONFLICT(episode_id) DO NOTHING`)
+    .run(episodeId, seriesId, showTitle, season, episode, oldFileId, reason);
+  return getRemediation(episodeId);
+}
+
+export function getRemediation(episodeId) {
+  return getDb().prepare('SELECT * FROM remediation_episodes WHERE episode_id = ?').get(episodeId) || null;
+}
+
+export function setRemediationState(episodeId, state, { details = null, attempts = null, blockedReleases = null } = {}) {
+  const db = getDb();
+  const row = getRemediation(episodeId);
+  if (!row) return;
+  db.prepare(`UPDATE remediation_episodes
+    SET state = ?, state_changed_at = strftime('%Y-%m-%dT%H:%M:%SZ','now'),
+        attempts = COALESCE(?, attempts), blocked_releases = COALESCE(?, blocked_releases),
+        details = COALESCE(?, details)
+    WHERE episode_id = ?`)
+    .run(state, attempts, blockedReleases, details, episodeId);
+}
+
+export function getActiveRemediations() {
+  return getDb().prepare(`SELECT * FROM remediation_episodes
+    WHERE state IN ('searching','verifying','importing') ORDER BY rowid`).all();
+}
+
+export function getNextQueuedRemediation() {
+  return getDb().prepare(`SELECT * FROM remediation_episodes WHERE state = 'queued' ORDER BY rowid LIMIT 1`).get() || null;
+}
+
+export function getRemediations() {
+  return getDb().prepare('SELECT * FROM remediation_episodes ORDER BY rowid DESC LIMIT 200').all();
+}
+
+export function resetRemediation(episodeId) {
+  getDb().prepare('DELETE FROM remediation_episodes WHERE episode_id = ?').run(episodeId);
+}

src/decoder.js

@@ -0,0 +1,74 @@
+import { execFile } from 'child_process';
+import { promisify } from 'util';
+import fs from 'fs';
+const execFileAsync = promisify(execFile);
+
+export function isIoTimeoutError(err) {
+  if (!err) return false;
+  if (err.killed && err.signal === 'SIGTERM') return true;   // child_process timeout
+  if (err.code === 'ETIMEDOUT' || err.code === 'EIO') return true;
+  return /timed? ?out|ETIMEDOUT|EIO/i.test(err.message || '');
+}
+
+async function decodeOnce(filePath, timeoutMs) {
+  try {
+    const { stderr } = await execFileAsync('ffmpeg', [
+      '-v', 'error',
+      '-xerror',
+      '-i', filePath,
+      '-map', '0:v?', '-map', '0:a?',
+      '-f', 'null', '-',
+    ], { timeout: timeoutMs, maxBuffer: 1024 * 1024 });
+    if (stderr && stderr.trim()) return { pass: false, error: stderr.trim().slice(0, 2000) };
+    return { pass: true, error: null };
+  } catch (err) {
+    return { pass: false, error: String(err.stderr || err.message).slice(0, 2000), ioTimeout: isIoTimeoutError(err) };
+  }
+}
+
+/**
+ * Full-stream decode. Retries once on timeout/IO errors (flaky disk, sleeping drive)
+ * before declaring failure; such retries are NOT corruption evidence.
+ */
+export async function fullDecode(filePath, { timeoutMs = 120000, retryIoOnce = true } = {}) {
+  try {
+    await fs.promises.access(filePath, fs.constants.R_OK);
+  } catch (err) {
+    return { pass: false, error: `File not found or unreadable: ${err.message}`, retried: false };
+  }
+  const first = await decodeOnce(filePath, timeoutMs);
+  if (first.pass) return { pass: true, error: null, retried: false };
+  if (retryIoOnce && first.ioTimeout) {
+    const second = await decodeOnce(filePath, timeoutMs);
+    return { pass: second.pass, error: second.error, retried: true };
+  }
+  return { pass: false, error: first.error, retried: false };
+}
+
+/**
+ * Confirms a file being written (unpackerr extraction) has stopped growing.
+ * Requires `checks` consecutive size-identical samples spaced intervalMs apart.
+ */
+export async function waitForStableFile(filePath, { intervalMs = 5000, checks = 2, maxWaitMs = 600000 } = {}) {
+  const start = Date.now();
+  let lastSize = -1;
+  let stable = 0;
+  while (Date.now() - start < maxWaitMs) {
+    let size;
+    try {
+      size = (await fs.promises.stat(filePath)).size;
+    } catch {
+      await new Promise(r => setTimeout(r, intervalMs));
+      continue; // not created yet (or vanished mid-extract)
+    }
+    if (size > 0 && size === lastSize) {
+      stable += 1;
+      if (stable >= checks) return true;
+    } else {
+      stable = 0;
+    }
+    lastSize = size;
+    await new Promise(r => setTimeout(r, intervalMs));
+  }
+  throw new Error(`waitForStableFile: ${filePath} did not stabilise within ${maxWaitMs}ms`);
+}

src/notifications.js

@@ -15,7 +15,10 @@ export async function sendAppriseNotification(eventType, { showTitle, season, ep
 
   const enabledRaw = getSetting('apprise_events') ?? 'episode_grabbed';
   const enabled = enabledRaw.split(',').map(s => s.trim()).filter(Boolean);
-  if (!enabled.includes(eventType)) return;
+  // 'remediation' whitelist entry matches any remediation_* event type
+  const allowed = enabled.includes(eventType) ||
+    (eventType.startsWith('remediation') && enabled.includes('remediation'));
+  if (!allowed) return;
 
   const ep = epLabel(season, episode);
   const show = showTitle ?? '';
@@ -27,6 +30,11 @@ export async function sendAppriseNotification(eventType, { showTitle, season, ep
     episode_remediated: `Episode Guard — Bad File Remediated`,
     episode_corrupt:    `Episode Guard — Corrupt File Warning`,
     error:              `Episode Guard — Error`,
+    remediation_started:        `Episode Guard — Corrupt Episode: Replacement Search Started`,
+    remediation_attempt:        `Episode Guard — Corrupt Episode: Replacement Failed Validation`,
+    remediation_swapped:        `Episode Guard — Corrupt Episode: Replaced With Verified Copy`,
+    remediation_exhausted:      `Episode Guard — Corrupt Episode: Gave Up After Max Releases`,
+    remediation_fallback:       `Episode Guard — Corrupt Episode: Delete-First Fallback Used`,
   };
 
   const bodies = {