diff --git a/backend/app/services/download_service.py b/backend/app/services/download_service.py
index 825bb82..979676d 100644
--- a/backend/app/services/download_service.py
+++ b/backend/app/services/download_service.py
@@ -423,6 +423,13 @@ class DownloadService:
await loop.run_in_executor(None, _upsert)
+ # Only now is it safe to call this walk's media seen: every file above
+ # has been through the importer. Had the run died before here they stay
+ # unmarked, and the next walk imports them from disk (ingest_core).
+ mark_seen = getattr(dl_result, "mark_seen_after_import", None)
+ if mark_seen is not None:
+ await loop.run_in_executor(None, mark_seen)
+
# #830 recapture: backfill source_filehash on EXISTING on-disk images so
# their post-body inline 
remaps to the local copy. A
# SEPARATE non-deleting channel (NOT the import list — that would unlink
diff --git a/backend/app/services/gallery_dl.py b/backend/app/services/gallery_dl.py
index b07fb1b..90933bf 100644
--- a/backend/app/services/gallery_dl.py
+++ b/backend/app/services/gallery_dl.py
@@ -17,6 +17,7 @@ import subprocess
import sys
import tempfile
import time
+from collections.abc import Callable
from dataclasses import dataclass, field
from datetime import UTC, datetime
from enum import StrEnum
@@ -230,6 +231,13 @@ class DownloadResult:
# the platform cooldown matches the hint instead of a flat default. None when
# unknown (no header, or not a rate-limit failure).
retry_after_seconds: float | None = None
+ # Native ingester only: marks this walk's fetched media seen in its ledger.
+ # Phase 3 calls it AFTER the import loop, never before — a file marked seen
+ # but not yet imported is invisible to every later walk, so a run killed in
+ # between orphaned it for good (TamadaHeijun's 12PCG post lost 7 of 13
+ # images to a stranded run, 2026-09-24). Unmarked, the next walk finds the
+ # file on disk with no ImageRecord and imports it. None on gallery-dl.
+ mark_seen_after_import: Callable[[], None] | None = None
def extract_errors_warnings(stderr: str) -> str:
diff --git a/backend/app/services/ingest_core.py b/backend/app/services/ingest_core.py
index 8105baf..8bb1e3d 100644
--- a/backend/app/services/ingest_core.py
+++ b/backend/app/services/ingest_core.py
@@ -36,6 +36,7 @@ from datetime import UTC, datetime, timedelta
from sqlalchemy import delete, func, select, text
from sqlalchemy.dialects.postgresql import insert as pg_insert
+from ..models import ImageRecord
from .gallery_dl import (
DownloadResult,
ErrorType,
@@ -261,6 +262,9 @@ class Ingester:
# source_filehash and (b) link the on-disk image to its Post (#1288) —
# WITHOUT re-downloading or unlinking the file. Empty outside recapture.
relink: list[tuple[str, str, str]] = []
+ # Media handed to phase 3 for import. Marked seen by phase 3 once the
+ # import has run (`mark_seen_after_import`), not here — see there.
+ fetched: list[tuple[str, str]] = []
downloaded = 0
errors = 0
quarantined = 0
@@ -313,6 +317,7 @@ class Ingester:
written_paths=written,
post_record_paths=list(post_records),
relink_source_paths=list(relink),
+ mark_seen_after_import=lambda: self._mark_seen(source_id, fetched),
stdout="\n".join(log_lines),
stderr="",
return_code=return_code,
@@ -512,6 +517,13 @@ class Ingester:
recapture=recapture,
)
+ # An on-disk file is only "done" if something imported it. One
+ # with no ImageRecord at its path was written by a run that died
+ # before phase 3 — it goes to import, not to the ledger.
+ imported_paths = self._recorded_paths([
+ str(o.path) for o in outcomes
+ if o.status == "skipped_disk" and o.path is not None
+ ])
to_mark: list[tuple[str, str]] = []
to_clear: list[str] = [] # recovered → drop any dead-letter row
to_fail: list[tuple[str, str, str]] = [] # (key, post_id, error)
@@ -523,11 +535,29 @@ class Ingester:
downloaded += 1
if outcome.path is not None:
written.append(str(outcome.path))
- to_mark.append((key, media_item.post_id))
+ fetched.append((key, media_item.post_id))
to_clear.append(key)
consecutive_seen = 0
+ elif (
+ outcome.status == "skipped_disk"
+ and outcome.path is not None
+ and str(outcome.path) not in imported_paths
+ ):
+ # On disk, never imported: a prior run wrote it and died
+ # before phase 3. Import it now. Safe to feed to
+ # attach_in_place because no record owns this path —
+ # the unlink below is about a file that IS the record.
+ written.append(str(outcome.path))
+ fetched.append((key, media_item.post_id))
+ to_clear.append(key)
+ skipped_count += 1
+ consecutive_seen += 1
+ log_lines.append(
+ f" post {media_item.post_id} — on disk but never "
+ f"imported: {outcome.path.name}"
+ )
elif outcome.status == "skipped_disk":
- # Already on disk (a prior run). Reconcile the ledger so a
+ # Already on disk and imported. Reconcile the ledger so a
# later tick skips it at tier-1 without a disk stat, but
# do NOT re-feed it to phase 3 — attach_in_place would see
# the duplicate sha256 and unlink the on-disk copy.
@@ -887,6 +917,16 @@ class Ingester:
)
session.commit()
+ def _recorded_paths(self, paths: list[str]) -> set[str]:
+ """Which of `paths` an ImageRecord already points at."""
+ if not paths:
+ return set()
+ with self.session_factory() as session:
+ rows = session.execute(
+ select(ImageRecord.path).where(ImageRecord.path.in_(paths))
+ ).scalars().all()
+ return set(rows)
+
def _mark_seen(self, source_id: int, items: list[tuple[str, str]]) -> None:
"""Idempotent upsert of (filehash, post_id) seen-ledger rows for a page.
diff --git a/frontend/src/components/subscriptions/SourceActions.vue b/frontend/src/components/subscriptions/SourceActions.vue
index e026aed..34e484f 100644
--- a/frontend/src/components/subscriptions/SourceActions.vue
+++ b/frontend/src/components/subscriptions/SourceActions.vue
@@ -45,8 +45,9 @@
>
Recapture post text & links
- Re-grab every post's body + external links and localize inline images
- already on disk — without re-downloading media
+ Re-grab every post's body + external links, localize inline images
+ already on disk, and import any downloaded file that never reached the
+ library — without re-downloading media
diff --git a/tests/test_download_service.py b/tests/test_download_service.py
index 7533dcb..a347646 100644
--- a/tests/test_download_service.py
+++ b/tests/test_download_service.py
@@ -9,9 +9,9 @@ from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock
import pytest
-from sqlalchemy import select
+from sqlalchemy import func, select
-from backend.app.models import Artist, DownloadEvent, ImportSettings, Source
+from backend.app.models import Artist, DownloadEvent, ImageRecord, ImportSettings, Source
from backend.app.services.credential_crypto import CredentialCrypto
from backend.app.services.credential_service import CredentialService
from backend.app.services.thumbnailer import Thumbnailer
@@ -152,6 +152,14 @@ async def test_download_source_attaches_written_files(
files_downloaded=2,
stdout=f"{f1}\n{f2}\n",
)
+ # The ledger is marked only once the files are in: a run killed before
+ # import must leave them unmarked so the next walk imports them.
+ imported_when_marked = []
+ result.mark_seen_after_import = lambda: imported_when_marked.append(
+ db_sync.execute(
+ select(func.count(ImageRecord.id)).where(ImageRecord.path.in_([str(f1), str(f2)]))
+ ).scalar_one()
+ )
fake_gdl = _fake_gdl_with_result(result)
sync_settings = db_sync.execute(
@@ -182,6 +190,7 @@ async def test_download_source_attaches_written_files(
assert ev.files_count == 2
assert ev.metadata_["import_summary"]["attached"] == 2
assert ev.metadata_["run_stats"]["downloaded_count"] == 2
+ assert imported_when_marked == [2]
@pytest.mark.asyncio
diff --git a/tests/test_patreon_ingester.py b/tests/test_patreon_ingester.py
index 774722b..7f56a96 100644
--- a/tests/test_patreon_ingester.py
+++ b/tests/test_patreon_ingester.py
@@ -7,6 +7,7 @@ real CDN. The ledger is real (a sync sessionmaker bound to the test engine), so
the tier-1 skip and the idempotent mark-seen run against actual rows.
"""
+import hashlib
from datetime import UTC, datetime, timedelta
import pytest
@@ -15,6 +16,7 @@ from sqlalchemy.orm import sessionmaker
from backend.app.models import (
Artist,
+ ImageRecord,
PatreonFailedMedia,
PatreonSeenMedia,
Source,
@@ -237,6 +239,9 @@ async def test_tick_downloads_unseen_and_marks_seen(source_id, sync_engine, tmp_
# plan #704: structured run_stats carry the real counts.
assert result.run_stats["downloaded_count"] == 2
assert result.posts_processed == 1
+ # The media wait for phase 3 to import them; only the post key is in yet.
+ assert _count_ledger(sync_engine, source_id) == 1
+ result.mark_seen_after_import()
# 2 media keys + 1 synthetic post key (body/links recaptured per post).
assert _count_ledger(sync_engine, source_id) == 3
# The post body + links are captured for media posts too (rides the walk).
@@ -263,6 +268,7 @@ async def test_quarantined_media_surfaced_in_result(source_id, sync_engine, tmp_
assert result.run_stats["quarantined_count"] == 1
assert result.run_stats["downloaded_count"] == 1
assert len(result.written_paths) == 1 # quarantined NOT written
+ result.mark_seen_after_import()
# Quarantined media is NOT marked seen (a fixed file may be re-fetched);
# m1 + the synthetic post key (body/links captured per post) = 2.
assert _count_ledger(sync_engine, source_id) == 2
@@ -573,6 +579,7 @@ async def test_recovery_tier2_disk_still_skips(source_id, sync_engine, tmp_path)
m1 = _media("p1", 1)
client = _FakeClient([(None, [("p1", [m1])])])
# File still on disk (a kept image) → tier-2 spares it even under recovery.
+ _seed_record(sync_engine, tmp_path / "p1_1.jpg")
downloader = _FakeDownloader(tmp_path, on_disk={_ledger_key(m1)})
ing = _ingester(sync_engine, tmp_path, client, downloader)
@@ -582,6 +589,7 @@ async def test_recovery_tier2_disk_still_skips(source_id, sync_engine, tmp_path)
)
assert result.files_downloaded == 0
assert downloader.download_calls == 0
+ assert result.written_paths == []
# Disk-skip reconciles the media key + the synthetic post key (recovery
# recaptures the body/links per post) = 2.
assert _count_ledger(sync_engine, source_id) == 2
@@ -609,6 +617,15 @@ async def test_backfill_recaptures_body_for_already_downloaded_post(
assert downloader.post_records == 1
+def _seed_record(sync_engine, path):
+ """An ImageRecord at `path` — the file was imported, not just downloaded."""
+ factory = sessionmaker(sync_engine, expire_on_commit=False)
+ with factory() as s:
+ s.add(ImageRecord(path=str(path), sha256=hashlib.sha256(str(path).encode()).hexdigest(),
+ size_bytes=1, mime="image/jpeg", origin="downloaded"))
+ s.commit()
+
+
def _seed_seen(sync_engine, source_id, key, post_id=None):
factory = sessionmaker(sync_engine, expire_on_commit=False)
with factory() as s:
@@ -629,6 +646,7 @@ async def test_backfill_skips_already_captured_post_but_recapture_forces_it(
# Pre-seed BOTH the media key and the synthetic post key as already seen.
_seed_seen(sync_engine, source_id, _ledger_key(m1), post_id="p1")
_seed_seen(sync_engine, source_id, "post:p1", post_id="p1")
+ _seed_record(sync_engine, tmp_path / "p1_1.jpg")
# 1) Plain backfill: post key is seen → gate skips body recapture.
client = _FakeClient([(None, [("p1", [m1])])])
@@ -696,6 +714,7 @@ async def test_gated_post_skipped_entirely_no_media_no_record(
assert downloader.download_calls == 1
assert len(result.post_record_paths) == 1 # only the open post
assert downloader.post_records == 1
+ result.mark_seen_after_import()
# The gated post left NO trace in the seen-ledger (no media key, no post key):
# only the open post's media key + synthetic post key are recorded.
assert _count_ledger(sync_engine, source_id) == 2
@@ -827,6 +846,57 @@ async def test_recapture_does_not_refetch_seen_media_missing_from_disk(
assert result.relink_source_paths == []
+# --- a run that dies between download and import ---------------------------
+# TamadaHeijun's 【12PCG】 post, 2026-09-24: 13 files on disk, 5 in the library.
+# A run wrote 01–08, marked them seen, and was killed before phase 3 imported
+# them; every later walk trusted the ledger and never looked again.
+
+
+@pytest.mark.asyncio
+async def test_a_run_that_dies_before_import_leaves_its_media_unmarked(
+ source_id, sync_engine, tmp_path,
+):
+ m1 = _media("p1", 1)
+ ing = _ingester(sync_engine, tmp_path, _FakeClient([(None, [("p1", [m1])])]),
+ _FakeDownloader(tmp_path))
+ ing.run(source_id=source_id, campaign_id="c1", artist_slug="ingest",
+ url="https://patreon.com/ingest", mode="tick")
+ # Phase 3 never ran, so `mark_seen_after_import` never did: only the post key.
+ assert _count_ledger(sync_engine, source_id) == 1
+
+ # The next walk finds the file on disk with no record, and imports it.
+ ing2 = _ingester(sync_engine, tmp_path, _FakeClient([(None, [("p1", [m1])])]),
+ _FakeDownloader(tmp_path, on_disk={_ledger_key(m1)}))
+ result = ing2.run(source_id=source_id, campaign_id="c1", artist_slug="ingest",
+ url="https://patreon.com/ingest", mode="tick")
+ assert result.written_paths == [str(tmp_path / "p1_1.jpg")]
+ assert result.files_downloaded == 0 # not fetched again
+ assert "on disk but never imported: p1_1.jpg" in result.stdout
+ result.mark_seen_after_import()
+ assert _count_ledger(sync_engine, source_id) == 2
+
+
+@pytest.mark.asyncio
+async def test_recapture_imports_a_seen_file_nothing_imported(
+ source_id, sync_engine, tmp_path,
+):
+ """The repair for files already orphaned: they are in the ledger, so only
+ a walk that looks past it — recapture — reaches them."""
+ m1, m2 = _media("p1", 1), _media("p1", 2)
+ for m in (m1, m2):
+ _seed_seen(sync_engine, source_id, _ledger_key(m), post_id="p1")
+ _seed_record(sync_engine, tmp_path / "p1_2.jpg") # m2 made it in; m1 did not
+ downloader = _FakeDownloader(tmp_path, on_disk={_ledger_key(m1), _ledger_key(m2)})
+ ing = _ingester(sync_engine, tmp_path, _FakeClient([(None, [("p1", [m1, m2])])]),
+ downloader)
+ result = ing.run(source_id=source_id, campaign_id="c1", artist_slug="ingest",
+ url="https://patreon.com/ingest", mode="recapture")
+
+ assert result.written_paths == [str(tmp_path / "p1_1.jpg")]
+ assert [r[0] for r in result.relink_source_paths] == [str(tmp_path / "p1_2.jpg")]
+ assert downloader.download_calls == 0
+
+
# --- dead-letter ledger (plan #705 #7) ------------------------------------