fix(renamer): safely replace higher revisions (#1078)

backend/src/module/api/downloader.py

@@ -1,7 +1,7 @@
 import logging
 import re
 
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel
 
 from module.database import Database
@@ -40,6 +40,42 @@ async def get_torrents():
         return await client.get_torrent_info(category="Bangumi", status_filter=None)
 
 
+@router.get("/rename-conflicts", dependencies=[Depends(get_current_user)])
+async def get_rename_conflicts():
+    """List durable media rename conflicts awaiting user action."""
+
+    async with Database() as db:
+        rows = await db.rename_operation.list_conflicts()
+    return [row.model_dump(mode="json") for row in rows]
+
+
+@router.post(
+    "/rename-conflicts/{operation_id}/retry",
+    dependencies=[Depends(get_current_user)],
+)
+async def retry_rename_conflict(operation_id: int):
+    """Clear one terminal conflict so the next rename pass revalidates it."""
+
+    async with Database() as db:
+        row = await db.rename_operation.get(operation_id)
+        if row is None:
+            raise HTTPException(status_code=404, detail="Rename conflict not found")
+        if row.state != "conflict" or row.kind != "conflict":
+            raise HTTPException(
+                status_code=409,
+                detail=(
+                    "Only non-destructive terminal conflicts can be retried; "
+                    "replacement recovery state must be preserved"
+                ),
+            )
+        await db.rename_operation.delete(operation_id)
+    return {
+        "status": True,
+        "msg_en": "Rename conflict cleared; it will be revalidated",
+        "msg_zh": "重命名冲突已清除，将在下一轮重新校验",
+    }
+
+
 @router.post("/torrents/pause", dependencies=[Depends(get_current_user)])
 async def pause_torrents(req: TorrentHashesRequest):
     hashes = "|".join(req.hashes)

backend/src/module/conf/const.py

@@ -31,6 +31,7 @@ DEFAULT_SETTINGS: dict[str, dict[str, Any]] = {
         "rename_method": "pn",
         "group_tag": False,
         "remove_bad_torrent": False,
+        "revision_conflict_policy": "hold",
     },
     "log": {
         "debug_enable": False,
@@ -112,6 +113,10 @@ ENV_TO_ATTR: dict[str, dict[str, Any]] = {
             "remove_bad_torrent",
             lambda e: e.lower() in ("true", "1", "t"),
         ),
+        "AB_REVISION_CONFLICT_POLICY": (
+            "revision_conflict_policy",
+            lambda e: e.lower(),
+        ),
     },
     "log": {
         "AB_DEBUG_MODE": ("debug_enable", lambda e: e.lower() in ("true", "1", "t")),

backend/src/module/core/loops.py

@@ -65,6 +65,9 @@ async def rename_tick(notifier: NotificationManager) -> None:
     async with DownloadClient() as client:
         renamer = Renamer(client)
         renamed_info = await renamer.rename()
+        rename_events = list(renamer.events)
+    if rename_events:
+        await asyncio.gather(*[notifier.send_event(event) for event in rename_events])
     if settings.notification.enable and renamed_info:
         # Gather across renamed items so a batch rename (e.g. first import of
         # a season) doesn't serialize N notification round-trips.

backend/src/module/database/aria2.py

@@ -6,6 +6,7 @@
 
 import json
 import logging
+from dataclasses import asdict, dataclass
 
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlmodel import select
@@ -15,6 +16,41 @@ from module.models.aria2 import Aria2Gid
 logger = logging.getLogger(__name__)
 
 
+@dataclass(frozen=True)
+class Aria2RenameIntent:
+    """Durable proof that this gid owned a filesystem move before it started."""
+
+    old_path: str
+    new_path: str
+    st_dev: int
+    st_ino: int
+    st_size: int
+    st_mtime_ns: int
+
+    def to_json(self) -> str:
+        return json.dumps(
+            {"version": 1, **asdict(self)}, ensure_ascii=False, sort_keys=True
+        )
+
+    @classmethod
+    def from_json(cls, raw: str | None) -> "Aria2RenameIntent | None":
+        if not raw:
+            return None
+        try:
+            data = json.loads(raw)
+        except (json.JSONDecodeError, TypeError):
+            return None
+        if not isinstance(data, dict) or data.get("version") != 1:
+            return None
+        path_fields = ("old_path", "new_path")
+        stat_fields = ("st_dev", "st_ino", "st_size", "st_mtime_ns")
+        if any(not isinstance(data.get(field), str) for field in path_fields):
+            return None
+        if any(type(data.get(field)) is not int for field in stat_fields):
+            return None
+        return cls(**{field: data[field] for field in (*path_fields, *stat_fields)})
+
+
 class Aria2GidDatabase:
     def __init__(self, session: AsyncSession):
         self.session = session
@@ -88,6 +124,7 @@ class Aria2GidDatabase:
                     category=old.category,
                     dedup_key=old.dedup_key,
                     renamed_paths=old.renamed_paths,
+                    rename_intent=old.rename_intent,
                     created_at=old.created_at,
                 )
             )
@@ -101,6 +138,8 @@ class Aria2GidDatabase:
             existing.renamed_paths = _merge_renamed_paths(
                 existing.renamed_paths, old.renamed_paths
             )
+            if existing.rename_intent is None:
+                existing.rename_intent = old.rename_intent
             self.session.add(existing)
         await self.session.delete(old)
         await self.session.commit()
@@ -126,6 +165,62 @@ class Aria2GidDatabase:
         mapping[old_path] = new_path
         await self.upsert(gid, renamed_paths=json.dumps(mapping, ensure_ascii=False))
 
+    async def get_rename_intent(self, gid: str) -> Aria2RenameIntent | None:
+        record = await self.session.get(Aria2Gid, gid)
+        if record is None or not record.rename_intent:
+            return None
+        intent = Aria2RenameIntent.from_json(record.rename_intent)
+        if intent is None:
+            logger.warning("Ignoring invalid rename_intent for gid %s", gid)
+        return intent
+
+    async def set_rename_intent(self, gid: str, intent: Aria2RenameIntent) -> None:
+        record = await self.session.get(Aria2Gid, gid)
+        if record is None:
+            record = Aria2Gid(gid=gid)
+        record.rename_intent = intent.to_json()
+        self.session.add(record)
+        await self.session.commit()
+
+    async def clear_rename_intent(
+        self,
+        gid: str,
+        expected: Aria2RenameIntent | None = None,
+    ) -> bool:
+        record = await self.session.get(Aria2Gid, gid)
+        if record is None:
+            return False
+        if (
+            expected is not None
+            and Aria2RenameIntent.from_json(record.rename_intent) != expected
+        ):
+            return False
+        record.rename_intent = None
+        self.session.add(record)
+        await self.session.commit()
+        return True
+
+    async def finalize_rename_intent(
+        self, gid: str, expected: Aria2RenameIntent
+    ) -> bool:
+        """Commit the sidecar mapping and clear its matching intent together."""
+        record = await self.session.get(Aria2Gid, gid)
+        if (
+            record is None
+            or Aria2RenameIntent.from_json(record.rename_intent) != expected
+        ):
+            return False
+        mapping = _decode_renamed_paths(record.renamed_paths)
+        for original_path, renamed_path in list(mapping.items()):
+            if renamed_path == expected.old_path:
+                mapping[original_path] = expected.new_path
+        mapping[expected.old_path] = expected.new_path
+        record.renamed_paths = json.dumps(mapping, ensure_ascii=False)
+        record.rename_intent = None
+        self.session.add(record)
+        await self.session.commit()
+        return True
+
     async def delete(self, gid: str) -> None:
         existing = await self.session.get(Aria2Gid, gid)
         if existing is not None:
@@ -147,3 +242,15 @@ def _merge_renamed_paths(left: str | None, right: str | None) -> str | None:
         return left
     merged.update(incoming)
     return json.dumps(merged, ensure_ascii=False)
+
+
+def _decode_renamed_paths(raw: str | None) -> dict[str, str]:
+    if not raw:
+        return {}
+    try:
+        data = json.loads(raw)
+    except json.JSONDecodeError:
+        return {}
+    if not isinstance(data, dict):
+        return {}
+    return {str(key): str(value) for key, value in data.items()}

backend/src/module/database/combine.py

@@ -20,6 +20,7 @@ from .migrations import (  # noqa: F401  (re-exported for existing importers)
     run_migrations_async,
 )
 from .movie import MovieDatabase
+from .rename_operation import RenameOperationDatabase
 from .rss import RSSDatabase
 from .torrent import TorrentDatabase
 from .user import UserDatabase
@@ -51,6 +52,7 @@ class Database:
         self.auth = AuthDatabase(self.session)
         self.inbox = InboxDatabase(self.session)
         self.llm_credential = LLMCredentialDatabase(self.session)
+        self.rename_operation = RenameOperationDatabase(self.session)
 
     async def __aenter__(self):
         return self

backend/src/module/database/migrations.py

@@ -17,7 +17,7 @@ from pydantic_core import PydanticUndefined
 from sqlalchemy import Connection, Engine, inspect, text
 from sqlmodel import SQLModel
 
-from module.models import ApiToken, AuthSession, Bangumi, Movie, User
+from module.models import ApiToken, AuthSession, Bangumi, Movie, RenameOperation, User
 from module.models.inbox import InboxMessage
 from module.models.llm_credential import LLMCredential
 from module.models.passkey import Passkey
@@ -38,6 +38,7 @@ TABLE_MODELS: list[type[SQLModel]] = [
     LLMCredential,
     AuthSession,
     ApiToken,
+    RenameOperation,
 ]
 
 # already_applied 守卫：接收 inspector，返回该迁移是否已生效
@@ -617,6 +618,169 @@ MIGRATIONS: tuple[Migration, ...] = (
             ),
         ),
     ),
+    Migration(
+        23,
+        "create durable rename/revision-replacement operation state",
+        (
+            """CREATE TABLE IF NOT EXISTS rename_operation (
+                id INTEGER PRIMARY KEY,
+                downloader_type VARCHAR(32) NOT NULL,
+                kind VARCHAR(32) NOT NULL DEFAULT 'conflict',
+                state VARCHAR(32) NOT NULL DEFAULT 'planned',
+                new_task_id VARCHAR NOT NULL,
+                old_task_id VARCHAR,
+                save_path VARCHAR NOT NULL,
+                source_path VARCHAR NOT NULL,
+                target_path VARCHAR NOT NULL,
+                staged_path VARCHAR,
+                bangumi_id INTEGER,
+                media_type VARCHAR(32),
+                season INTEGER,
+                episode REAL,
+                group_name VARCHAR,
+                resolution VARCHAR(32),
+                old_revision INTEGER,
+                new_revision INTEGER,
+                revision_metadata TEXT,
+                attempt_count INTEGER NOT NULL DEFAULT 0,
+                retry_at TIMESTAMP,
+                lease_owner VARCHAR(64),
+                lease_expires_at TIMESTAMP,
+                notified_at TIMESTAMP,
+                last_error TEXT,
+                created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                CONSTRAINT ck_rename_operation_state CHECK (
+                    state IN ('conflict', 'retry', 'running', 'planned', 'old_staged',
+                              'new_promoted', 'old_removed', 'done')
+                ),
+                CONSTRAINT ck_rename_operation_attempt_count CHECK (
+                    attempt_count >= 0
+                )
+            )""",
+            "CREATE UNIQUE INDEX IF NOT EXISTS ux_rename_operation_identity "
+            "ON rename_operation(downloader_type, new_task_id, save_path, "
+            "source_path, target_path)",
+            "CREATE UNIQUE INDEX IF NOT EXISTS ux_rename_operation_active_target "
+            "ON rename_operation(downloader_type, save_path, target_path) "
+            "WHERE state NOT IN ('done')",
+            "CREATE INDEX IF NOT EXISTS ix_rename_operation_state_retry_at "
+            "ON rename_operation(state, retry_at)",
+            "CREATE INDEX IF NOT EXISTS ix_rename_operation_new_task_id "
+            "ON rename_operation(new_task_id)",
+            "CREATE INDEX IF NOT EXISTS ix_rename_operation_old_task_id "
+            "ON rename_operation(old_task_id)",
+        ),
+        all_checks(
+            table_exists("rename_operation"),
+            index_exists("rename_operation", "ux_rename_operation_identity"),
+            index_exists("rename_operation", "ux_rename_operation_active_target"),
+            index_exists("rename_operation", "ix_rename_operation_state_retry_at"),
+            index_exists("rename_operation", "ix_rename_operation_new_task_id"),
+            index_exists("rename_operation", "ix_rename_operation_old_task_id"),
+        ),
+        (
+            (
+                """CREATE TABLE IF NOT EXISTS rename_operation (
+                    id INTEGER PRIMARY KEY,
+                    downloader_type VARCHAR(32) NOT NULL,
+                    kind VARCHAR(32) NOT NULL DEFAULT 'conflict',
+                    state VARCHAR(32) NOT NULL DEFAULT 'planned',
+                    new_task_id VARCHAR NOT NULL,
+                    old_task_id VARCHAR,
+                    save_path VARCHAR NOT NULL,
+                    source_path VARCHAR NOT NULL,
+                    target_path VARCHAR NOT NULL,
+                    staged_path VARCHAR,
+                    bangumi_id INTEGER,
+                    media_type VARCHAR(32),
+                    season INTEGER,
+                    episode REAL,
+                    group_name VARCHAR,
+                    resolution VARCHAR(32),
+                    old_revision INTEGER,
+                    new_revision INTEGER,
+                    revision_metadata TEXT,
+                    attempt_count INTEGER NOT NULL DEFAULT 0,
+                    retry_at TIMESTAMP,
+                    lease_owner VARCHAR(64),
+                    lease_expires_at TIMESTAMP,
+                    notified_at TIMESTAMP,
+                    last_error TEXT,
+                    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                    CONSTRAINT ck_rename_operation_state CHECK (
+                        state IN ('conflict', 'retry', 'running', 'planned', 'old_staged',
+                                  'new_promoted', 'old_removed', 'done')
+                    ),
+                    CONSTRAINT ck_rename_operation_attempt_count CHECK (
+                        attempt_count >= 0
+                    )
+                )""",
+                table_exists("rename_operation"),
+            ),
+            (
+                "CREATE UNIQUE INDEX IF NOT EXISTS ux_rename_operation_identity "
+                "ON rename_operation(downloader_type, new_task_id, save_path, "
+                "source_path, target_path)",
+                index_exists("rename_operation", "ux_rename_operation_identity"),
+            ),
+            (
+                "CREATE UNIQUE INDEX IF NOT EXISTS ux_rename_operation_active_target "
+                "ON rename_operation(downloader_type, save_path, target_path) "
+                "WHERE state NOT IN ('done')",
+                index_exists("rename_operation", "ux_rename_operation_active_target"),
+            ),
+            (
+                "CREATE INDEX IF NOT EXISTS ix_rename_operation_state_retry_at "
+                "ON rename_operation(state, retry_at)",
+                index_exists("rename_operation", "ix_rename_operation_state_retry_at"),
+            ),
+            (
+                "CREATE INDEX IF NOT EXISTS ix_rename_operation_new_task_id "
+                "ON rename_operation(new_task_id)",
+                index_exists("rename_operation", "ix_rename_operation_new_task_id"),
+            ),
+            (
+                "CREATE INDEX IF NOT EXISTS ix_rename_operation_old_task_id "
+                "ON rename_operation(old_task_id)",
+                index_exists("rename_operation", "ix_rename_operation_old_task_id"),
+            ),
+        ),
+    ),
+    Migration(
+        24,
+        "add durable filesystem rename intent to aria2 gid state",
+        (
+            """CREATE TABLE IF NOT EXISTS aria2_gid (
+                gid VARCHAR NOT NULL PRIMARY KEY,
+                bangumi_id INTEGER REFERENCES bangumi(id),
+                category VARCHAR,
+                dedup_key VARCHAR,
+                renamed_paths TEXT DEFAULT NULL,
+                created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+            )""",
+            "ALTER TABLE aria2_gid ADD COLUMN rename_intent TEXT DEFAULT NULL",
+        ),
+        column_exists("aria2_gid", "rename_intent"),
+        (
+            (
+                """CREATE TABLE IF NOT EXISTS aria2_gid (
+                    gid VARCHAR NOT NULL PRIMARY KEY,
+                    bangumi_id INTEGER REFERENCES bangumi(id),
+                    category VARCHAR,
+                    dedup_key VARCHAR,
+                    renamed_paths TEXT DEFAULT NULL,
+                    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+                )""",
+                table_exists("aria2_gid"),
+            ),
+            (
+                "ALTER TABLE aria2_gid ADD COLUMN rename_intent TEXT DEFAULT NULL",
+                column_exists("aria2_gid", "rename_intent"),
+            ),
+        ),
+    ),
 )
 
 # 由迁移列表派生，新增迁移时无需手动同步

backend/src/module/database/rename_operation.py

@@ -0,0 +1,419 @@
+"""Repository for durable rename/revision-replacement operations (#1078)."""
+
+from datetime import datetime, timedelta
+
+from sqlalchemy import delete, or_, update
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlmodel import col, select
+
+from module.models.rename_operation import (
+    RENAME_OPERATION_STATES,
+    RenameOperation,
+    RenameOperationState,
+    utc_now,
+)
+
+
+def _validate_state(state: str) -> None:
+    if state not in RENAME_OPERATION_STATES:
+        raise ValueError(f"Unsupported rename operation state: {state}")
+
+
+class RenameOperationDatabase:
+    def __init__(self, session: AsyncSession):
+        self.session = session
+
+    async def get(self, operation_id: int | None) -> RenameOperation | None:
+        if operation_id is None:
+            return None
+        return await self.session.get(RenameOperation, operation_id)
+
+    async def get_by_identity(
+        self,
+        *,
+        downloader_type: str,
+        new_task_id: str,
+        save_path: str,
+        source_path: str,
+        target_path: str,
+    ) -> RenameOperation | None:
+        result = await self.session.execute(
+            select(RenameOperation).where(
+                RenameOperation.downloader_type == downloader_type,
+                RenameOperation.new_task_id == new_task_id,
+                RenameOperation.save_path == save_path,
+                RenameOperation.source_path == source_path,
+                RenameOperation.target_path == target_path,
+            )
+        )
+        return result.scalar_one_or_none()
+
+    async def get_by_target(
+        self,
+        *,
+        downloader_type: str,
+        save_path: str,
+        target_path: str,
+        active_only: bool = True,
+    ) -> RenameOperation | None:
+        statement = select(RenameOperation).where(
+            RenameOperation.downloader_type == downloader_type,
+            RenameOperation.save_path == save_path,
+            RenameOperation.target_path == target_path,
+        )
+        if active_only:
+            statement = statement.where(RenameOperation.state != "done")
+        statement = statement.order_by(  # type: ignore[assignment]
+            col(RenameOperation.updated_at).desc()
+        )
+        result = await self.session.execute(statement)
+        return result.scalars().first()
+
+    async def get_or_create(
+        self, operation: RenameOperation
+    ) -> tuple[RenameOperation, bool]:
+        """Return the identity row, creating it when absent.
+
+        The pre-read handles the normal idempotent path.  The database unique
+        indexes remain authoritative for concurrent writers; an identity race
+        is re-read after rollback, while an active-target collision is exposed
+        as ``IntegrityError`` for the caller to reconcile.
+        """
+
+        existing = await self.get_by_identity(
+            downloader_type=operation.downloader_type,
+            new_task_id=operation.new_task_id,
+            save_path=operation.save_path,
+            source_path=operation.source_path,
+            target_path=operation.target_path,
+        )
+        if existing is not None:
+            return existing, False
+
+        operation.created_at = operation.created_at or utc_now()
+        operation.updated_at = utc_now()
+        try:
+            # Keep an active-target race inside a SAVEPOINT.  Rolling back the
+            # whole session would expire unrelated rows already returned to the
+            # caller and make a recoverable uniqueness conflict poison the
+            # surrounding operation.
+            async with self.session.begin_nested():
+                self.session.add(operation)
+                await self.session.flush()
+        except IntegrityError:
+            existing = await self.get_by_identity(
+                downloader_type=operation.downloader_type,
+                new_task_id=operation.new_task_id,
+                save_path=operation.save_path,
+                source_path=operation.source_path,
+                target_path=operation.target_path,
+            )
+            if existing is not None:
+                return existing, False
+            raise
+        await self.session.commit()
+        await self.session.refresh(operation)
+        return operation, True
+
+    async def upsert_conflict(
+        self, operation: RenameOperation
+    ) -> tuple[RenameOperation, bool]:
+        """Persist a terminal hold conflict without duplicating notifications."""
+
+        operation.state = "conflict"
+        operation.retry_at = None
+        existing = await self.get_by_identity(
+            downloader_type=operation.downloader_type,
+            new_task_id=operation.new_task_id,
+            save_path=operation.save_path,
+            source_path=operation.source_path,
+            target_path=operation.target_path,
+        )
+        if existing is None:
+            return await self.get_or_create(operation)
+
+        # Refresh reconciliation metadata without resetting attempts,
+        # notification history, or the original creation timestamp.
+        for field_name in (
+            "kind",
+            "old_task_id",
+            "staged_path",
+            "bangumi_id",
+            "media_type",
+            "season",
+            "episode",
+            "group_name",
+            "resolution",
+            "old_revision",
+            "new_revision",
+            "revision_metadata",
+            "last_error",
+        ):
+            setattr(existing, field_name, getattr(operation, field_name))
+        existing.state = "conflict"
+        existing.retry_at = None
+        existing.updated_at = utc_now()
+        self.session.add(existing)
+        await self.session.commit()
+        await self.session.refresh(existing)
+        return existing, False
+
+    async def set_state(
+        self,
+        operation_id: int | None,
+        state: RenameOperationState,
+        *,
+        retry_at: datetime | None = None,
+        last_error: str | None = None,
+    ) -> RenameOperation | None:
+        _validate_state(state)
+        row = await self.get(operation_id)
+        if row is None:
+            return None
+        row.state = state
+        row.retry_at = retry_at
+        row.last_error = last_error
+        row.updated_at = utc_now()
+        self.session.add(row)
+        await self.session.commit()
+        await self.session.refresh(row)
+        return row
+
+    async def claim(
+        self,
+        operation_id: int | None,
+        *,
+        from_states: tuple[RenameOperationState, ...],
+        to_state: RenameOperationState,
+        now: datetime | None = None,
+    ) -> RenameOperation | None:
+        """Atomically claim a due operation and count one execution attempt."""
+
+        if operation_id is None or not from_states:
+            return None
+        for state in from_states:
+            _validate_state(state)
+        _validate_state(to_state)
+        claimed_at = now or utc_now()
+        result = await self.session.execute(
+            update(RenameOperation)
+            .where(
+                col(RenameOperation.id) == operation_id,
+                col(RenameOperation.state).in_(from_states),
+                or_(
+                    col(RenameOperation.retry_at).is_(None),
+                    col(RenameOperation.retry_at) <= claimed_at,
+                ),
+            )
+            .values(
+                state=to_state,
+                attempt_count=RenameOperation.attempt_count + 1,
+                retry_at=None,
+                updated_at=claimed_at,
+            )
+            .execution_options(synchronize_session="fetch")
+        )
+        await self.session.commit()
+        if not result.rowcount:  # type: ignore[attr-defined]
+            return None
+        row = await self.get(operation_id)
+        if row is not None:
+            await self.session.refresh(row)
+        return row
+
+    async def recover_stale_running(
+        self, operation_id: int | None, *, before: datetime
+    ) -> bool:
+        """CAS a crashed ordinary rename lease back to retryable state."""
+
+        if operation_id is None:
+            return False
+        now = utc_now()
+        result = await self.session.execute(
+            update(RenameOperation)
+            .where(
+                col(RenameOperation.id) == operation_id,
+                col(RenameOperation.state) == "running",
+                col(RenameOperation.updated_at) <= before,
+            )
+            .values(state="retry", retry_at=None, updated_at=now)
+            .execution_options(synchronize_session="fetch")
+        )
+        await self.session.commit()
+        return bool(result.rowcount)  # type: ignore[attr-defined]
+
+    async def claim_replacement_lease(
+        self,
+        operation_id: int | None,
+        *,
+        owner: str,
+        now: datetime | None = None,
+        lease_for: timedelta = timedelta(minutes=5),
+    ) -> RenameOperation | None:
+        """Fence one replacement Saga across processes for a bounded lease."""
+
+        if operation_id is None:
+            return None
+        claimed_at = now or utc_now()
+        result = await self.session.execute(
+            update(RenameOperation)
+            .where(
+                col(RenameOperation.id) == operation_id,
+                col(RenameOperation.kind) == "replacement",
+                col(RenameOperation.state).in_(
+                    ("planned", "old_staged", "new_promoted", "old_removed")
+                ),
+                or_(
+                    col(RenameOperation.lease_owner).is_(None),
+                    col(RenameOperation.lease_expires_at).is_(None),
+                    col(RenameOperation.lease_expires_at) <= claimed_at,
+                ),
+            )
+            .values(
+                lease_owner=owner,
+                lease_expires_at=claimed_at + lease_for,
+                attempt_count=RenameOperation.attempt_count + 1,
+                updated_at=claimed_at,
+            )
+            .execution_options(synchronize_session="fetch")
+        )
+        await self.session.commit()
+        if not result.rowcount:  # type: ignore[attr-defined]
+            return None
+        return await self.get(operation_id)
+
+    async def set_state_claimed(
+        self,
+        operation_id: int | None,
+        *,
+        owner: str,
+        state: RenameOperationState,
+        retry_at: datetime | None = None,
+        last_error: str | None = None,
+    ) -> RenameOperation | None:
+        """Advance a replacement only while the caller still owns its lease."""
+
+        if operation_id is None:
+            return None
+        _validate_state(state)
+        now = utc_now()
+        result = await self.session.execute(
+            update(RenameOperation)
+            .where(
+                col(RenameOperation.id) == operation_id,
+                col(RenameOperation.lease_owner) == owner,
+                col(RenameOperation.lease_expires_at) > now,
+            )
+            .values(
+                state=state,
+                retry_at=retry_at,
+                last_error=last_error,
+                lease_owner=None,
+                lease_expires_at=None,
+                updated_at=now,
+            )
+            .execution_options(synchronize_session="fetch")
+        )
+        await self.session.commit()
+        if not result.rowcount:  # type: ignore[attr-defined]
+            return None
+        return await self.get(operation_id)
+
+    async def release_replacement_lease(
+        self, operation_id: int | None, *, owner: str
+    ) -> bool:
+        if operation_id is None:
+            return False
+        result = await self.session.execute(
+            update(RenameOperation)
+            .where(
+                col(RenameOperation.id) == operation_id,
+                col(RenameOperation.lease_owner) == owner,
+            )
+            .values(lease_owner=None, lease_expires_at=None, updated_at=utc_now())
+            .execution_options(synchronize_session="fetch")
+        )
+        await self.session.commit()
+        return bool(result.rowcount)  # type: ignore[attr-defined]
+
+    async def mark_notified(
+        self, operation_id: int | None, when: datetime | None = None
+    ) -> bool:
+        """Set ``notified_at`` once; concurrent or repeated calls return False."""
+
+        if operation_id is None:
+            return False
+        notified_at = when or utc_now()
+        result = await self.session.execute(
+            update(RenameOperation)
+            .where(
+                col(RenameOperation.id) == operation_id,
+                col(RenameOperation.notified_at).is_(None),
+            )
+            .values(notified_at=notified_at, updated_at=notified_at)
+            .execution_options(synchronize_session="fetch")
+        )
+        await self.session.commit()
+        return bool(result.rowcount)  # type: ignore[attr-defined]
+
+    async def list_conflicts(self, limit: int = 100) -> list[RenameOperation]:
+        result = await self.session.execute(
+            select(RenameOperation)
+            .where(RenameOperation.state == "conflict")
+            .order_by(col(RenameOperation.updated_at).desc())
+            .limit(limit)
+        )
+        return list(result.scalars().all())
+
+    async def list_retryable(
+        self, now: datetime | None = None, limit: int = 100
+    ) -> list[RenameOperation]:
+        ready_at = now or utc_now()
+        result = await self.session.execute(
+            select(RenameOperation)
+            .where(
+                RenameOperation.state == "retry",
+                or_(
+                    col(RenameOperation.retry_at).is_(None),
+                    col(RenameOperation.retry_at) <= ready_at,
+                ),
+            )
+            .order_by(col(RenameOperation.retry_at).asc())
+            .limit(limit)
+        )
+        return list(result.scalars().all())
+
+    async def list_active_replacements(self, limit: int = 100) -> list[RenameOperation]:
+        result = await self.session.execute(
+            select(RenameOperation)
+            .where(
+                RenameOperation.kind == "replacement",
+                col(RenameOperation.state).in_(
+                    ("planned", "old_staged", "new_promoted", "old_removed")
+                ),
+            )
+            .order_by(col(RenameOperation.updated_at).asc())
+            .limit(limit)
+        )
+        return list(result.scalars().all())
+
+    async def delete(self, operation_id: int | None) -> bool:
+        row = await self.get(operation_id)
+        if row is None:
+            return False
+        await self.session.delete(row)
+        await self.session.commit()
+        return True
+
+    async def prune_done(self, before: datetime) -> int:
+        result = await self.session.execute(
+            delete(RenameOperation)
+            .where(
+                col(RenameOperation.state) == "done",
+                col(RenameOperation.updated_at) < before,
+            )
+            .execution_options(synchronize_session=False)
+        )
+        await self.session.commit()
+        return int(result.rowcount or 0)  # type: ignore[attr-defined]

backend/src/module/downloader/__init__.py

@@ -3,6 +3,8 @@ from .base import (
     CoreDownloaderClient,
     DownloaderCapabilities,
     DownloaderClient,
+    RenameOutcome,
+    RenameResult,
 )
 from .download_client import DownloadClient, shutdown
 
@@ -12,5 +14,7 @@ __all__ = [
     "DownloadClient",
     "DownloaderCapabilities",
     "DownloaderClient",
+    "RenameOutcome",
+    "RenameResult",
     "shutdown",
 ]

backend/src/module/downloader/base.py

@@ -20,6 +20,38 @@ class AddResult(Enum):
     FAILED = "failed"
 
 
+class RenameOutcome(str, Enum):
+    """Downloader-independent outcome of a single file rename."""
+
+    RENAMED = "renamed"
+    ALREADY_APPLIED = "already_applied"
+    DESTINATION_EXISTS = "destination_exists"
+    RETRYABLE_FAILURE = "retryable_failure"
+
+
+@dataclass(frozen=True)
+class RenameResult:
+    """Structured rename result preserved across the downloader facade.
+
+    ``bool(result)`` deliberately keeps the old success/failure behaviour while
+    callers migrate to inspecting :attr:`outcome`.  It must never make a
+    collision or retryable failure truthy.
+    """
+
+    outcome: RenameOutcome
+    detail: str | None = None
+
+    @property
+    def succeeded(self) -> bool:
+        return self.outcome in {
+            RenameOutcome.RENAMED,
+            RenameOutcome.ALREADY_APPLIED,
+        }
+
+    def __bool__(self) -> bool:
+        return self.succeeded
+
+
 @dataclass(frozen=True)
 class DownloaderCapabilities:
     """What a concrete download client can do.
@@ -83,6 +115,8 @@ class DownloaderClient(Protocol):
 
     async def torrents_info(self, status_filter, category, tag=None) -> list[dict]: ...
 
+    async def torrent_exists(self, torrent_hash: str) -> bool | None: ...
+
     async def torrents_files(self, torrent_hash: str) -> list[dict]: ...
 
     async def torrents_delete(self, hash, delete_files: bool = True) -> bool: ...
@@ -93,7 +127,7 @@ class DownloaderClient(Protocol):
 
     async def torrents_rename_file(
         self, torrent_hash, old_path, new_path, verify: bool = True
-    ) -> bool: ...
+    ) -> RenameResult: ...
 
     async def move_torrent(self, hashes, new_location) -> None: ...
 

backend/src/module/downloader/client/aria2_downloader.py

@@ -8,7 +8,13 @@ from typing import ClassVar
 import httpx
 
 from module.database import Database
-from module.downloader.base import AddResult, DownloaderCapabilities
+from module.database.aria2 import Aria2RenameIntent
+from module.downloader.base import (
+    AddResult,
+    DownloaderCapabilities,
+    RenameOutcome,
+    RenameResult,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -220,6 +226,30 @@ class Aria2Downloader:
         os.makedirs(os.path.dirname(new_abs), exist_ok=True)
         shutil.move(old_abs, new_abs)
 
+    @staticmethod
+    def _rename_intent(
+        old_path: str, new_path: str, source_stat: os.stat_result
+    ) -> Aria2RenameIntent:
+        return Aria2RenameIntent(
+            old_path=old_path,
+            new_path=new_path,
+            st_dev=source_stat.st_dev,
+            st_ino=source_stat.st_ino,
+            st_size=source_stat.st_size,
+            st_mtime_ns=source_stat.st_mtime_ns,
+        )
+
+    @staticmethod
+    def _intent_matches_stat(
+        intent: Aria2RenameIntent, target_stat: os.stat_result
+    ) -> bool:
+        return (
+            intent.st_dev == target_stat.st_dev
+            and intent.st_ino == target_stat.st_ino
+            and intent.st_size == target_stat.st_size
+            and intent.st_mtime_ns == target_stat.st_mtime_ns
+        )
+
     @staticmethod
     def _move_files(files: list[dict], old_dir: str, new_dir: str) -> None:
         old_dir = os.path.normpath(old_dir)
@@ -430,7 +460,7 @@ class Aria2Downloader:
             stopped = await self._call("tellStopped", [0, 1000]) or []
         except (Aria2RpcError, Aria2ConnectionError) as e:
             logger.error("Failed to query downloads: %s", e)
-            return []
+            raise
         raw = [*active, *waiting, *stopped]
         followed_by: dict[str, str] = {}
         for d in raw:
@@ -507,6 +537,20 @@ class Aria2Downloader:
             )
         return result
 
+    async def torrent_exists(self, gid: str) -> bool | None:
+        """Return False only when aria2 explicitly confirms a gid is absent."""
+        try:
+            await self._call("tellStatus", [gid, ["status"]])
+        except Aria2RpcError as e:
+            if self._is_not_found_error(e):
+                return False
+            logger.warning("Cannot confirm whether aria2 gid %s exists: %s", gid, e)
+            return None
+        except Aria2ConnectionError as e:
+            logger.warning("Cannot confirm whether aria2 gid %s exists: %s", gid, e)
+            return None
+        return True
+
     async def torrents_files(self, torrent_hash: str) -> list[dict]:
         try:
             files = await self._call("getFiles", [torrent_hash])
@@ -535,7 +579,7 @@ class Aria2Downloader:
 
     async def torrents_rename_file(
         self, torrent_hash, old_path, new_path, verify: bool = True
-    ) -> bool:
+    ) -> RenameResult:
         try:
             status = await self._call("tellStatus", [torrent_hash, ["dir"]])
         except (Aria2RpcError, Aria2ConnectionError) as e:
@@ -544,32 +588,118 @@ class Aria2Downloader:
                 torrent_hash,
                 e,
             )
-            return False
+            return RenameResult(RenameOutcome.RETRYABLE_FAILURE, detail=str(e))
         save_dir = (status or {}).get("dir", "")
         if not save_dir:
-            return False
+            return RenameResult(
+                RenameOutcome.RETRYABLE_FAILURE,
+                detail="aria2 returned no download directory",
+            )
         old_abs = os.path.join(save_dir, old_path)
         new_abs = os.path.join(save_dir, new_path)
+        old_exists, new_exists = await asyncio.gather(
+            asyncio.to_thread(os.path.exists, old_abs),
+            asyncio.to_thread(os.path.exists, new_abs),
+        )
+        async with Database() as db:
+            persisted_intent = await db.aria2.get_rename_intent(torrent_hash)
+
+        if not old_exists:
+            matching_intent = (
+                persisted_intent
+                if persisted_intent is not None
+                and persisted_intent.old_path == old_path
+                and persisted_intent.new_path == new_path
+                else None
+            )
+            if not new_exists:
+                if matching_intent is not None:
+                    async with Database() as db:
+                        await db.aria2.clear_rename_intent(
+                            torrent_hash, matching_intent
+                        )
+                return RenameResult(
+                    RenameOutcome.RETRYABLE_FAILURE,
+                    detail=f"rename source is missing: {old_abs}",
+                )
+            if matching_intent is None:
+                return RenameResult(
+                    RenameOutcome.DESTINATION_EXISTS,
+                    detail=(
+                        "destination exists without a matching durable rename "
+                        f"intent: {new_abs}"
+                    ),
+                )
+            try:
+                target_stat = await asyncio.to_thread(os.stat, new_abs)
+            except OSError as e:
+                return RenameResult(RenameOutcome.RETRYABLE_FAILURE, detail=str(e))
+            if not self._intent_matches_stat(matching_intent, target_stat):
+                async with Database() as db:
+                    await db.aria2.clear_rename_intent(torrent_hash, matching_intent)
+                return RenameResult(
+                    RenameOutcome.DESTINATION_EXISTS,
+                    detail=f"destination does not match durable rename intent: {new_abs}",
+                )
+            async with Database() as db:
+                finalized = await db.aria2.finalize_rename_intent(
+                    torrent_hash, matching_intent
+                )
+            if not finalized:
+                return RenameResult(
+                    RenameOutcome.RETRYABLE_FAILURE,
+                    detail="durable rename intent changed before recovery commit",
+                )
+            return RenameResult(RenameOutcome.ALREADY_APPLIED)
+
+        try:
+            source_stat = await asyncio.to_thread(os.stat, old_abs)
+        except OSError as e:
+            return RenameResult(RenameOutcome.RETRYABLE_FAILURE, detail=str(e))
+        intent = self._rename_intent(old_path, new_path, source_stat)
+        async with Database() as db:
+            await db.aria2.set_rename_intent(torrent_hash, intent)
         try:
             await asyncio.to_thread(self._move_file, old_abs, new_abs)
         except FileExistsError:
+            async with Database() as db:
+                await db.aria2.clear_rename_intent(torrent_hash, intent)
             logger.warning(
                 "Refusing to overwrite existing file %s, rename of %s skipped",
                 new_abs,
                 old_abs,
             )
-            return False
+            return RenameResult(
+                RenameOutcome.DESTINATION_EXISTS,
+                detail=f"destination already exists: {new_abs}",
+            )
         except OSError as e:
+            async with Database() as db:
+                await db.aria2.clear_rename_intent(torrent_hash, intent)
             logger.warning("Failed to rename file %s -> %s: %s", old_abs, new_abs, e)
-            return False
-        if verify:
-            exists = await asyncio.to_thread(os.path.exists, new_abs)
-            if not exists:
-                logger.debug("Rename reported success but %s is missing", new_abs)
-                return False
+            return RenameResult(RenameOutcome.RETRYABLE_FAILURE, detail=str(e))
+        try:
+            target_stat = await asyncio.to_thread(os.stat, new_abs)
+        except OSError as e:
+            logger.debug("Rename reported success but %s cannot be verified", new_abs)
+            # Keep the intent: the move may have completed and a later retry can
+            # still reconcile it if the target becomes visible.
+            return RenameResult(RenameOutcome.RETRYABLE_FAILURE, detail=str(e))
+        if not self._intent_matches_stat(intent, target_stat):
+            async with Database() as db:
+                await db.aria2.clear_rename_intent(torrent_hash, intent)
+            return RenameResult(
+                RenameOutcome.DESTINATION_EXISTS,
+                detail=f"renamed target identity changed unexpectedly: {new_abs}",
+            )
         async with Database() as db:
-            await db.aria2.set_renamed_path(torrent_hash, old_path, new_path)
-        return True
+            finalized = await db.aria2.finalize_rename_intent(torrent_hash, intent)
+        if not finalized:
+            return RenameResult(
+                RenameOutcome.RETRYABLE_FAILURE,
+                detail="durable rename intent changed before sidecar commit",
+            )
+        return RenameResult(RenameOutcome.RENAMED)
 
     # ------------------------------------------------------------------
     # Management
@@ -580,40 +710,98 @@ class Aria2Downloader:
         ok = True
         async with Database() as db:
             for gid in gids:
-                save_dir = None
-                try:
-                    status = await self._call("tellStatus", [gid, ["dir"]])
-                    save_dir = (status or {}).get("dir")
-                except (Aria2RpcError, Aria2ConnectionError) as e:
-                    logger.debug("Could not resolve dir for %s: %s", gid, e)
-
-                if delete_files and save_dir:
+                if delete_files:
                     try:
+                        status = await self._call("tellStatus", [gid, ["dir"]])
+                        save_dir = (status or {}).get("dir")
+                        if not save_dir:
+                            raise Aria2ConnectionError(
+                                f"aria2 returned no download directory for {gid}"
+                            )
                         files = await self._call("getFiles", [gid])
-                    except (Aria2RpcError, Aria2ConnectionError):
-                        files = []
+                    except Aria2RpcError as e:
+                        if self._is_not_found_error(e):
+                            # The task disappeared after a previous successful
+                            # cleanup (or was removed externally).  There is no
+                            # downloader-owned file list left to resolve.
+                            await db.aria2.delete(gid)
+                            continue
+                        logger.warning(
+                            "Cannot safely delete current files for %s: %s", gid, e
+                        )
+                        ok = False
+                        continue
+                    except Aria2ConnectionError as e:
+                        # If the current paths cannot be enumerated safely, do
+                        # not remove the task or its sidecar.  A later retry can
+                        # still reconcile the staged files.
+                        logger.warning(
+                            "Cannot safely delete current files for %s: %s", gid, e
+                        )
+                        ok = False
+                        continue
+
+                    renamed_paths = await db.aria2.get_renamed_paths(gid)
+                    files_removed = True
                     for f in files or []:
-                        path = f.get("path")
-                        if not path:
+                        raw_path = f.get("path")
+                        if not raw_path:
+                            continue
+                        relative_path = os.path.relpath(raw_path, save_dir)
+                        current_relative_path = self._translate_renamed_path(
+                            relative_path, renamed_paths
+                        )
+                        current_path = os.path.normpath(
+                            os.path.join(save_dir, current_relative_path)
+                        )
+                        boundary = os.path.normpath(save_dir)
+                        try:
+                            within_boundary = (
+                                os.path.commonpath([boundary, current_path]) == boundary
+                            )
+                        except ValueError:
+                            within_boundary = False
+                        if not within_boundary:
+                            logger.warning(
+                                "Refusing to delete aria2 path outside save dir: %s",
+                                current_path,
+                            )
+                            files_removed = False
                             continue
                         try:
                             await asyncio.to_thread(
-                                self._remove_file_and_empty_dirs, path, save_dir
+                                self._remove_file_and_empty_dirs,
+                                current_path,
+                                save_dir,
                             )
                         except OSError as e:
-                            logger.warning("Failed to delete file %s: %s", path, e)
+                            logger.warning(
+                                "Failed to delete current file %s: %s", current_path, e
+                            )
+                            files_removed = False
+                    if not files_removed:
+                        ok = False
+                        continue
 
+                removed = False
                 try:
                     await self._call("forceRemove", [gid])
+                    removed = True
                 except Aria2RpcError as e:
-                    if not self._is_not_found_error(e):
+                    if self._is_not_found_error(e):
+                        removed = True
+                    else:
                         logger.error("Failed to remove %s: %s", gid, e)
                         ok = False
                 except Aria2ConnectionError as e:
                     logger.error("Failed to remove %s: %s", gid, e)
                     ok = False
 
-                await db.aria2.delete(gid)
+                # The local mapping is recovery state.  Keep it whenever aria2
+                # may still own the task; delete only after confirmed removal
+                # or an explicit not-found response.
+                if removed:
+                    await db.aria2.delete(gid)
         return ok
 
     async def torrents_pause(self, hashes: str | list[str]) -> None:

backend/src/module/downloader/client/mock_downloader.py

@@ -8,7 +8,12 @@ qBittorrent instance. All operations return success and log their actions.
 import logging
 from typing import Any, ClassVar
 
-from module.downloader.base import AddResult, DownloaderCapabilities
+from module.downloader.base import (
+    AddResult,
+    DownloaderCapabilities,
+    RenameOutcome,
+    RenameResult,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -92,6 +97,9 @@ class MockDownloader:
         logger.debug("get_torrents_by_tag(%s)", tag)
         return [t for t in self._torrents.values() if tag in t.get("tags", [])]
 
+    async def torrent_exists(self, torrent_hash: str) -> bool | None:
+        return torrent_hash in self._torrents
+
     async def torrents_files(self, torrent_hash: str) -> list[dict]:
         """Return files for a torrent."""
         logger.debug("torrents_files(%s)", torrent_hash)
@@ -159,9 +167,9 @@ class MockDownloader:
 
     async def torrents_rename_file(
         self, torrent_hash: str, old_path: str, new_path: str, verify: bool = True
-    ) -> bool:
+    ) -> RenameResult:
         logger.info(f"rename: {old_path} -> {new_path}")
-        return True
+        return RenameResult(RenameOutcome.RENAMED)
 
     async def rss_add_feed(self, url: str, item_path: str):
         self._feeds[item_path] = {"url": url, "path": item_path}

backend/src/module/downloader/client/qb_downloader.py

@@ -8,7 +8,12 @@ from typing import ClassVar
 import httpx
 
 from module.ab_decorator import qb_connect_failed_wait
-from module.downloader.base import AddResult, DownloaderCapabilities
+from module.downloader.base import (
+    AddResult,
+    DownloaderCapabilities,
+    RenameOutcome,
+    RenameResult,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -311,6 +316,28 @@ class QbDownloader:
             ]
         return torrents
 
+    async def torrent_exists(self, torrent_hash: str) -> bool | None:
+        """Query one hash without confusing a failed request with absence."""
+
+        try:
+            resp = await self._get("torrents/info", params={"hashes": torrent_hash})
+            if resp.status_code >= 300:
+                logger.warning(
+                    "Could not verify qBittorrent task %s: HTTP %s",
+                    torrent_hash,
+                    resp.status_code,
+                )
+                return None
+            payload = resp.json()
+        except (ConnectionError, OSError, ValueError, httpx.RequestError) as e:
+            logger.warning("Could not verify qBittorrent task %s: %s", torrent_hash, e)
+            return None
+        return isinstance(payload, list) and any(
+            str(item.get("hash", "")).lower() == torrent_hash.lower()
+            for item in payload
+            if isinstance(item, dict)
+        )
+
     @qb_connect_failed_wait
     async def torrents_files(self, torrent_hash: str):
         resp = await self._get("torrents/files", params={"hash": torrent_hash})
@@ -555,7 +582,7 @@ class QbDownloader:
 
     async def torrents_rename_file(
         self, torrent_hash, old_path, new_path, verify: bool = True
-    ) -> bool:
+    ) -> RenameResult:
         try:
             resp = await self._post(
                 "torrents/renameFile",
@@ -563,13 +590,19 @@ class QbDownloader:
             )
             if resp.status_code == 409:
                 logger.debug("Conflict409Error: %s >> %s", old_path, new_path)
-                return False
+                return RenameResult(
+                    RenameOutcome.DESTINATION_EXISTS,
+                    detail="qBittorrent rejected rename with HTTP 409",
+                )
             # qB 5.2 对成功的 renameFile 回 204（空响应体全局改为 204）
             if resp.status_code >= 300:
-                return False
+                return RenameResult(
+                    RenameOutcome.RETRYABLE_FAILURE,
+                    detail=f"qBittorrent rename returned HTTP {resp.status_code}",
+                )
 
             if not verify:
-                return True
+                return RenameResult(RenameOutcome.RENAMED)
 
             # Verify the rename actually happened by checking file list
             # qBittorrent can return 200 but delay the actual rename (e.g., while seeding)
@@ -579,7 +612,7 @@ class QbDownloader:
                 await asyncio.sleep(delay)
                 files = await self.torrents_files(torrent_hash)
                 if any(f.get("name") == new_path for f in files):
-                    return True
+                    return RenameResult(RenameOutcome.RENAMED)
                 # new_path 未出现（旧名仍在，或两个名字都不在列表里）一律不算
                 # 成功：盲目返回 True 会让 renamer 每周期重复改名 + 发通知
                 # (#754/#749)。返回 False 由调用方走 _PENDING_RENAME_COOLDOWN 退避。
@@ -590,10 +623,13 @@ class QbDownloader:
                         new_path,
                         old_path,
                     )
-            return False
+            return RenameResult(
+                RenameOutcome.RETRYABLE_FAILURE,
+                detail="qBittorrent did not expose the target path after rename",
+            )
         except (httpx.ConnectError, httpx.RequestError, httpx.TimeoutException) as e:
             logger.warning(f"Failed to rename file {old_path}: {e}")
-            return False
+            return RenameResult(RenameOutcome.RETRYABLE_FAILURE, detail=str(e))
 
     async def rss_add_feed(self, url, item_path):
         resp = await self._post(

backend/src/module/downloader/download_client.py

@@ -7,7 +7,7 @@ from module.conf import settings
 from module.models import Bangumi, Torrent
 from module.network import RequestContent
 
-from .base import AddResult, DownloaderClient
+from .base import AddResult, DownloaderClient, RenameOutcome, RenameResult
 from .path import gen_save_path
 
 logger = logging.getLogger(__name__)
@@ -312,18 +312,51 @@ class DownloadClient:
             return []
         return await self.client.torrents_files(torrent_hash=torrent_hash)
 
+    async def torrent_exists(self, torrent_hash: str) -> bool | None:
+        """Return task existence, or ``None`` when the backend cannot prove it.
+
+        Destructive recovery paths must distinguish a confirmed absence from
+        a transient query failure; an empty bulk snapshot is not sufficient.
+        """
+
+        if not self._supports("can_query", "torrent_exists"):
+            return None
+        return await self.client.torrent_exists(torrent_hash)
+
     async def rename_torrent_file(
         self, _hash, old_path, new_path, verify: bool = True
-    ) -> bool:
+    ) -> RenameResult:
         if not self._supports("can_rename", "rename_torrent_file"):
-            return False
-        result = await self.client.torrents_rename_file(
+            return RenameResult(
+                RenameOutcome.RETRYABLE_FAILURE,
+                detail=f"{type(self.client).__name__} does not support file rename",
+            )
+        raw_result = await self.client.torrents_rename_file(
             torrent_hash=_hash, old_path=old_path, new_path=new_path, verify=verify
         )
-        if result:
+        # Compatibility for test doubles and third-party clients that still
+        # implement the pre-3.3.4 boolean contract.  Concrete built-in clients
+        # all return RenameResult, and the facade always exposes RenameResult.
+        if isinstance(raw_result, RenameResult):
+            result = raw_result
+        else:
+            result = RenameResult(
+                (
+                    RenameOutcome.RENAMED
+                    if raw_result
+                    else RenameOutcome.RETRYABLE_FAILURE
+                ),
+                detail=None if raw_result else "legacy downloader returned false",
+            )
+        if result.succeeded:
             logger.info(f"{old_path} >> {new_path}")
         else:
-            logger.debug("Rename failed: %s >> %s", old_path, new_path)
+            logger.debug(
+                "Rename %s: %s >> %s",
+                result.outcome.value,
+                old_path,
+                new_path,
+            )
         return result
 
     async def delete_torrent(self, hashes, delete_files: bool = True) -> bool:

backend/src/module/manager/revision_policy.py

@@ -0,0 +1,105 @@
+"""Admission helpers for destructive higher-revision replacement (#1078)."""
+
+from __future__ import annotations
+
+import re
+import unicodedata
+from dataclasses import dataclass
+from pathlib import PurePosixPath
+
+from module.parser.analyser.selector import parse_configured_release_title
+from module.parser.analyser.tokenizer import MediaType
+from module.parser.release_policy import preference_identity, preference_revision
+
+
+@dataclass(frozen=True, slots=True)
+class RevisionIdentity:
+    bangumi_id: int
+    media_type: MediaType
+    season: int
+    episode: int | float
+    group: str
+    resolution: str
+    revision: int
+
+
+def _normalize_label(value: str | None) -> str:
+    if not value:
+        return ""
+    normalized = unicodedata.normalize("NFKC", value).casefold()
+    return re.sub(r"[^\w]+", "", normalized)
+
+
+def _adjust_episode(episode: int | float, offset: int) -> int | float:
+    if episode == 0 and offset:
+        return 0
+    adjusted = episode + offset
+    if adjusted < 0 or (adjusted == 0 and episode > 0):
+        return episode
+    return adjusted
+
+
+def parse_revision_identity(
+    torrent_name: str,
+    *,
+    bangumi_id: int | None,
+    default_season: int,
+    episode_offset: int = 0,
+) -> RevisionIdentity | None:
+    """Return the strict identity required before an old file may be deleted."""
+
+    if bangumi_id is None:
+        return None
+    release = parse_configured_release_title(torrent_name)
+    if release is None:
+        return None
+    identity = preference_identity(release, default_season=default_season)
+    if identity is None:
+        return None
+    media_type, season, episode = identity
+    group = _normalize_label(release.group)
+    resolution = _normalize_label(release.resolution)
+    if not group or not resolution:
+        return None
+    return RevisionIdentity(
+        bangumi_id=bangumi_id,
+        media_type=media_type,
+        season=season,
+        episode=_adjust_episode(episode, episode_offset),
+        group=group,
+        resolution=resolution,
+        revision=preference_revision(release),
+    )
+
+
+def is_strict_upgrade(old: RevisionIdentity, new: RevisionIdentity) -> bool:
+    """Whether *new* may destructively replace *old*."""
+
+    return bool(same_release_identity(old, new) and new.revision > old.revision)
+
+
+def same_release_identity(old: RevisionIdentity, new: RevisionIdentity) -> bool:
+    """Whether two revisions describe the same strictly matched resource."""
+
+    return bool(
+        old.bangumi_id == new.bangumi_id
+        and old.media_type is new.media_type
+        and old.season == new.season
+        and old.episode == new.episode
+        and old.group == new.group
+        and old.resolution == new.resolution
+    )
+
+
+def replacement_staged_path(
+    target_path: str, *, old_task_id: str, old_revision: int
+) -> str:
+    """Build a deterministic same-directory temporary path for saga recovery."""
+
+    normalized = target_path.replace("\\", "/")
+    path = PurePosixPath(normalized)
+    suffix = path.suffix
+    stem = path.name[: -len(suffix)] if suffix else path.name
+    staged_name = f"{stem}.ab-replaced-v{old_revision}-{old_task_id[:8]}{suffix}"
+    parent = str(path.parent)
+    return staged_name if parent == "." else f"{parent}/{staged_name}"

backend/src/module/models/__init__.py

@@ -5,6 +5,11 @@ from .inbox import InboxMessage
 from .llm_credential import LLMCredential
 from .movie import Movie, MovieUpdate
 from .passkey import Passkey, PasskeyCreate, PasskeyDelete, PasskeyList
+from .rename_operation import (
+    RENAME_OPERATION_STATES,
+    RenameOperation,
+    RenameOperationState,
+)
 from .response import APIResponse, ResponseModel
 from .rss import RSSItem, RSSUpdate
 from .torrent import EpisodeFile, SubtitleFile, Torrent, TorrentUpdate

backend/src/module/models/aria2.py

@@ -21,4 +21,5 @@ class Aria2Gid(SQLModel, table=True):
     category: Optional[str] = None
     dedup_key: Optional[str] = Field(default=None, index=True)
     renamed_paths: Optional[str] = None
+    rename_intent: Optional[str] = None
     created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

backend/src/module/models/config.py

@@ -71,6 +71,10 @@ class BangumiManage(BaseModel):
     rename_method: str = Field(default="pn", description="Rename method")
     group_tag: bool = Field(default=False, description="Enable group tag")
     remove_bad_torrent: bool = Field(default=False, description="Remove bad torrent")
+    revision_conflict_policy: Literal["hold", "replace"] = Field(
+        default="hold",
+        description="How to handle a higher revision targeting an existing episode",
+    )
     # 关闭后 refresh_rss 不再把未匹配种子入库（孤儿记录）；代价是这些条目
     # 每轮会被重新内存匹配（廉价），好处是后补规则能立即接住仍在源里的旧集
     track_orphans: bool = Field(

backend/src/module/models/rename_operation.py

@@ -0,0 +1,122 @@
+"""Durable rename/revision-replacement operation state (#1078).
+
+Downloader calls deliberately happen outside database transactions.  This row
+is the small, durable state machine that lets a later rename pass reconcile the
+observable downloader state and continue an idempotent operation after a retry
+or process restart.
+"""
+
+from datetime import datetime, timezone
+from typing import Literal
+
+from sqlalchemy import CheckConstraint, Index, text
+from sqlmodel import Field, SQLModel
+
+RenameOperationState = Literal[
+    "conflict",
+    "retry",
+    "running",
+    "planned",
+    "old_staged",
+    "new_promoted",
+    "old_removed",
+    "done",
+]
+
+RENAME_OPERATION_STATES: tuple[str, ...] = (
+    "conflict",
+    "retry",
+    "running",
+    "planned",
+    "old_staged",
+    "new_promoted",
+    "old_removed",
+    "done",
+)
+
+
+def utc_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+class RenameOperation(SQLModel, table=True):
+    """One conflict or recoverable revision-replacement operation.
+
+    The full identity prevents duplicate rows for one downloader task/path
+    request.  A partial unique index separately reserves an active canonical
+    target; completed history is excluded so a later V3 operation is not
+    blocked by an older V2 row.
+    """
+
+    __tablename__ = "rename_operation"
+    __table_args__ = (
+        CheckConstraint(
+            "state IN ('conflict', 'retry', 'running', 'planned', 'old_staged', "
+            "'new_promoted', 'old_removed', 'done')",
+            name="ck_rename_operation_state",
+        ),
+        CheckConstraint(
+            "attempt_count >= 0",
+            name="ck_rename_operation_attempt_count",
+        ),
+        Index(
+            "ux_rename_operation_identity",
+            "downloader_type",
+            "new_task_id",
+            "save_path",
+            "source_path",
+            "target_path",
+            unique=True,
+        ),
+        Index(
+            "ux_rename_operation_active_target",
+            "downloader_type",
+            "save_path",
+            "target_path",
+            unique=True,
+            sqlite_where=text("state NOT IN ('done')"),
+            postgresql_where=text("state NOT IN ('done')"),
+        ),
+        Index("ix_rename_operation_state_retry_at", "state", "retry_at"),
+        Index("ix_rename_operation_new_task_id", "new_task_id"),
+        Index("ix_rename_operation_old_task_id", "old_task_id"),
+    )
+
+    id: int | None = Field(default=None, primary_key=True)
+    downloader_type: str = Field(max_length=32)
+    kind: str = Field(default="conflict", max_length=32)
+    state: str = Field(default="planned", max_length=32)
+
+    # Downloader task ownership.  A hold conflict may not have a resolved old
+    # owner yet; every operation always has the incoming/new task id.
+    new_task_id: str
+    old_task_id: str | None = None
+
+    # All paths are downloader-relative except save_path, which is the
+    # normalized task root.  staged_path is deterministic for crash recovery.
+    save_path: str
+    source_path: str
+    target_path: str
+    staged_path: str | None = None
+
+    # Parsed content identity and revision snapshot used to revalidate a
+    # replacement after restart.  revision_metadata stores the lossless JSON
+    # snapshot/raw titles for fields that do not warrant schema columns.
+    bangumi_id: int | None = None
+    media_type: str | None = Field(default=None, max_length=32)
+    season: int | None = None
+    episode: float | None = None
+    group_name: str | None = None
+    resolution: str | None = Field(default=None, max_length=32)
+    old_revision: int | None = None
+    new_revision: int | None = None
+    revision_metadata: str | None = None
+
+    attempt_count: int = 0
+    retry_at: datetime | None = None
+    lease_owner: str | None = Field(default=None, max_length=64)
+    lease_expires_at: datetime | None = None
+    notified_at: datetime | None = None
+    last_error: str | None = None
+    created_at: datetime = Field(default_factory=utc_now)
+    updated_at: datetime = Field(default_factory=utc_now)

backend/src/module/notification/__init__.py

@@ -5,6 +5,7 @@ from .events import (
     LLMAuthFailureEvent,
     LLMPluginInstallFailedEvent,
     OffsetReviewEvent,
+    RenameConflictEvent,
     RssFailureEvent,
     SystemEvent,
     UpdateAppliedEvent,
@@ -21,6 +22,7 @@ __all__ = [
     "NotificationManager",
     "NotificationProvider",
     "OffsetReviewEvent",
+    "RenameConflictEvent",
     "PROVIDER_REGISTRY",
     "RssFailureEvent",
     "SystemEvent",

backend/src/module/notification/events.py

@@ -248,6 +248,38 @@ class LLMPluginInstallFailedEvent:
         )
 
 
+@dataclass(slots=True, frozen=True)
+class RenameConflictEvent:
+    """A media rename reached a durable target-path conflict."""
+
+    kind: ClassVar[str] = "rename_conflict"
+    severity: ClassVar[str] = "warning"
+    once: ClassVar[bool] = False
+
+    task_id: str
+    torrent_name: str
+    target_path: str
+    reason: str
+
+    def dedup_key(self) -> Optional[str]:
+        return f"rename_conflict:{self.task_id}:{self.target_path}"
+
+    def payload(self) -> dict:
+        return {
+            "task_id": self.task_id,
+            "torrent_name": self.torrent_name,
+            "target_path": self.target_path,
+            "reason": self.reason,
+        }
+
+    def describe(self) -> tuple[str, str]:
+        return (
+            "媒体文件重命名冲突",
+            f"种子：{self.torrent_name}\n目标：{self.target_path}\n"
+            f"原因：{self.reason}",
+        )
+
+
 # 除「新集数」以外的通知事件的联合类型。
 SystemEvent = (
     RssFailureEvent
@@ -258,4 +290,5 @@ SystemEvent = (
     | UpdateAppliedEvent
     | LLMAuthFailureEvent
     | LLMPluginInstallFailedEvent
+    | RenameConflictEvent
 )

backend/src/test/conftest.py

@@ -125,15 +125,16 @@ def mock_settings(test_settings):
 @pytest.fixture
 def mock_qb_client():
     """Mock QbDownloader that simulates qBittorrent API responses."""
-    from module.downloader import AddResult
+    from module.downloader import AddResult, RenameOutcome, RenameResult
 
     client = AsyncMock()
     client.auth.return_value = True
     client.logout.return_value = None
     client.check_host.return_value = True
     client.torrents_info.return_value = []
+    client.torrent_exists.return_value = False
     client.torrents_files.return_value = []
-    client.torrents_rename_file.return_value = True
+    client.torrents_rename_file.return_value = RenameResult(RenameOutcome.RENAMED)
     client.add_torrents.return_value = AddResult.ADDED
     client.torrents_delete.return_value = None
     client.torrents_pause.return_value = None

backend/src/test/test_api_downloader.py

@@ -7,6 +7,7 @@ from fastapi import FastAPI
 from fastapi.testclient import TestClient
 
 from module.api import v1
+from module.models import RenameOperation
 from module.security.api import get_current_user
 
 # ---------------------------------------------------------------------------
@@ -77,6 +78,76 @@ class TestAuthRequired:
         response = unauthed_client.get("/api/v1/downloader/torrents")
         assert response.status_code == 401
 
+
+class TestRenameConflicts:
+    def test_list_conflicts(self, authed_client):
+        row = RenameOperation(
+            id=7,
+            downloader_type="qbittorrent",
+            kind="conflict",
+            state="conflict",
+            new_task_id="new-v2",
+            save_path="/downloads/Show/Season 1",
+            source_path="raw-v2.mkv",
+            target_path="Show S01E01.mkv",
+            last_error="target exists",
+        )
+        with patch("module.api.downloader.Database") as mock_database:
+            db = MagicMock()
+            db.rename_operation.list_conflicts = AsyncMock(return_value=[row])
+            mock_database.return_value.__aenter__ = AsyncMock(return_value=db)
+            mock_database.return_value.__aexit__ = AsyncMock(return_value=False)
+            response = authed_client.get("/api/v1/downloader/rename-conflicts")
+
+        assert response.status_code == 200
+        assert response.json()[0]["id"] == 7
+        assert response.json()[0]["state"] == "conflict"
+
+    def test_retry_deletes_terminal_conflict(self, authed_client):
+        row = RenameOperation(
+            id=7,
+            downloader_type="qbittorrent",
+            kind="conflict",
+            state="conflict",
+            new_task_id="new-v2",
+            save_path="/downloads/Show/Season 1",
+            source_path="raw-v2.mkv",
+            target_path="Show S01E01.mkv",
+        )
+        with patch("module.api.downloader.Database") as mock_database:
+            db = MagicMock()
+            db.rename_operation.get = AsyncMock(return_value=row)
+            db.rename_operation.delete = AsyncMock(return_value=True)
+            mock_database.return_value.__aenter__ = AsyncMock(return_value=db)
+            mock_database.return_value.__aexit__ = AsyncMock(return_value=False)
+            response = authed_client.post("/api/v1/downloader/rename-conflicts/7/retry")
+
+        assert response.status_code == 200
+        db.rename_operation.delete.assert_awaited_once_with(7)
+
+    def test_retry_rejects_active_replacement(self, authed_client):
+        row = RenameOperation(
+            id=7,
+            downloader_type="qbittorrent",
+            kind="replacement",
+            state="new_promoted",
+            new_task_id="new-v2",
+            old_task_id="old-v1",
+            save_path="/downloads/Show/Season 1",
+            source_path="raw-v2.mkv",
+            target_path="Show S01E01.mkv",
+        )
+        with patch("module.api.downloader.Database") as mock_database:
+            db = MagicMock()
+            db.rename_operation.get = AsyncMock(return_value=row)
+            mock_database.return_value.__aenter__ = AsyncMock(return_value=db)
+            mock_database.return_value.__aexit__ = AsyncMock(return_value=False)
+            response = authed_client.post("/api/v1/downloader/rename-conflicts/7/retry")
+
+        assert response.status_code == 409
+
+
+class TestOtherAuthRequired:
     @patch("module.security.api.DEV_AUTH_BYPASS", False)
     def test_pause_torrents_unauthorized(self, unauthed_client):
         """POST /downloader/torrents/pause without auth returns 401."""

backend/src/test/test_aria2_downloader.py

@@ -15,7 +15,8 @@ import httpx
 import pytest
 
 from module.database import Database
-from module.downloader import AddResult
+from module.database.aria2 import Aria2RenameIntent
+from module.downloader import AddResult, RenameOutcome
 from module.downloader.client.aria2_downloader import (
     Aria2ConnectionError,
     Aria2Downloader,
@@ -27,6 +28,18 @@ def _aria2() -> Aria2Downloader:
     return Aria2Downloader("http://localhost:6800", "u", "secret-token")
 
 
+def _rename_intent(old_path: str, new_path: str, source) -> Aria2RenameIntent:
+    stat_result = source.stat()
+    return Aria2RenameIntent(
+        old_path=old_path,
+        new_path=new_path,
+        st_dev=stat_result.st_dev,
+        st_ino=stat_result.st_ino,
+        st_size=stat_result.st_size,
+        st_mtime_ns=stat_result.st_mtime_ns,
+    )
+
+
 def _rpc_response(result=None, error=None):
     resp = MagicMock()
     body: dict = {"jsonrpc": "2.0", "id": 1}
@@ -648,13 +661,47 @@ class TestTorrentsInfo:
         assert result[0]["tags"] == ""
         assert result[0]["category"] == ""
 
-    async def test_query_failure_returns_empty_list(self):
+    async def test_connection_query_failure_is_raised(self):
         aria2 = _aria2()
         with patch.object(
             aria2, "_call", AsyncMock(side_effect=Aria2ConnectionError("down"))
         ):
-            result = await aria2.torrents_info(status_filter=None, category=None)
-        assert result == []
+            with pytest.raises(Aria2ConnectionError, match="down"):
+                await aria2.torrents_info(status_filter=None, category=None)
+
+    async def test_rpc_query_failure_is_raised(self):
+        aria2 = _aria2()
+        with patch.object(
+            aria2, "_call", AsyncMock(side_effect=Aria2RpcError(1, "failed"))
+        ):
+            with pytest.raises(Aria2RpcError, match="failed"):
+                await aria2.torrents_info(status_filter=None, category=None)
+
+    async def test_torrent_exists_returns_true_for_known_gid(self):
+        aria2 = _aria2()
+        with patch.object(
+            aria2, "_call", AsyncMock(return_value={"status": "complete"})
+        ) as call:
+            assert await aria2.torrent_exists("gidA") is True
+        call.assert_awaited_once_with("tellStatus", ["gidA", ["status"]])
+
+    async def test_torrent_exists_returns_false_only_for_explicit_not_found(self):
+        aria2 = _aria2()
+        with patch.object(
+            aria2,
+            "_call",
+            AsyncMock(side_effect=Aria2RpcError(1, "GID#gidA is not found")),
+        ):
+            assert await aria2.torrent_exists("gidA") is False
+
+    @pytest.mark.parametrize(
+        "error",
+        [Aria2ConnectionError("down"), Aria2RpcError(1, "Internal error")],
+    )
+    async def test_torrent_exists_returns_none_when_presence_is_unknown(self, error):
+        aria2 = _aria2()
+        with patch.object(aria2, "_call", AsyncMock(side_effect=error)):
+            assert await aria2.torrent_exists("gidA") is None
 
     async def test_torrents_info_zero_total_length_progress_is_zero(self):
         """磁力链在拉元数据阶段 totalLength 是字符串 "0"（truthy）——
@@ -802,6 +849,67 @@ class TestTorrentsInfo:
             assert await db.aria2.get("payload-gid") is not None
 
 
+# ---------------------------------------------------------------------------
+# aria2_gid durable rename intent repository
+# ---------------------------------------------------------------------------
+
+
+class TestRenameIntentDatabase:
+    async def test_persists_and_reads_typed_rename_intent(self, tmp_path):
+        source = tmp_path / "old.mkv"
+        source.write_bytes(b"data")
+        intent = _rename_intent("old.mkv", "new.mkv", source)
+
+        async with Database() as db:
+            await db.aria2.set_rename_intent("gidA", intent)
+            assert await db.aria2.get_rename_intent("gidA") == intent
+            record = await db.aria2.get("gidA")
+
+        assert record is not None
+        assert record.rename_intent is not None
+
+    async def test_finalize_updates_mapping_and_clears_intent_atomically(
+        self, tmp_path
+    ):
+        source = tmp_path / "old.mkv"
+        source.write_bytes(b"data")
+        intent = _rename_intent("old.mkv", "new.mkv", source)
+
+        async with Database() as db:
+            await db.aria2.set_rename_intent("gidA", intent)
+            finalized = await db.aria2.finalize_rename_intent("gidA", intent)
+
+            assert finalized is True
+            assert await db.aria2.get_renamed_paths("gidA") == {"old.mkv": "new.mkv"}
+            assert await db.aria2.get_rename_intent("gidA") is None
+            record = await db.aria2.get("gidA")
+            assert record is not None
+            assert record.rename_intent is None
+
+    async def test_finalize_rejects_stale_intent_without_mutating_mapping(
+        self, tmp_path
+    ):
+        source = tmp_path / "old.mkv"
+        source.write_bytes(b"data")
+        persisted = _rename_intent("old.mkv", "new.mkv", source)
+        stale = Aria2RenameIntent(
+            old_path="old.mkv",
+            new_path="other.mkv",
+            st_dev=persisted.st_dev,
+            st_ino=persisted.st_ino,
+            st_size=persisted.st_size,
+            st_mtime_ns=persisted.st_mtime_ns,
+        )
+
+        async with Database() as db:
+            await db.aria2.set_rename_intent("gidA", persisted)
+            finalized = await db.aria2.finalize_rename_intent("gidA", stale)
+
+            assert finalized is False
+            assert await db.aria2.get_renamed_paths("gidA") == {}
+            assert await db.aria2.get_rename_intent("gidA") == persisted
+
+
 # ---------------------------------------------------------------------------
 # torrents_files: relative-path mapping
 # ---------------------------------------------------------------------------
@@ -872,7 +980,7 @@ class TestTorrentsRenameFile:
         with patch.object(aria2, "_call", AsyncMock(side_effect=fake_call)):
             result = await aria2.torrents_rename_file("gidA", "old.mkv", "new.mkv")
 
-        assert result is True
+        assert result.outcome is RenameOutcome.RENAMED
         assert not (tmp_path / "old.mkv").exists()
         assert (tmp_path / "new.mkv").read_bytes() == b"data"
 
@@ -887,7 +995,8 @@ class TestTorrentsRenameFile:
             return {"dir": str(tmp_path)}
 
         with patch.object(aria2, "_call", AsyncMock(side_effect=fake_call)):
-            assert await aria2.torrents_rename_file("gidA", "old.mkv", "new.mkv")
+            result = await aria2.torrents_rename_file("gidA", "old.mkv", "new.mkv")
+            assert result.outcome is RenameOutcome.RENAMED
             files = await aria2.torrents_files("gidA")
 
         assert files == [{"name": "new.mkv", "size": 4}]
@@ -903,8 +1012,10 @@ class TestTorrentsRenameFile:
             return {"dir": str(tmp_path)}
 
         with patch.object(aria2, "_call", AsyncMock(side_effect=fake_call)):
-            assert await aria2.torrents_rename_file("gidA", "old.mkv", "mid.mkv")
-            assert await aria2.torrents_rename_file("gidA", "mid.mkv", "new.mkv")
+            first = await aria2.torrents_rename_file("gidA", "old.mkv", "mid.mkv")
+            second = await aria2.torrents_rename_file("gidA", "mid.mkv", "new.mkv")
+            assert first.outcome is RenameOutcome.RENAMED
+            assert second.outcome is RenameOutcome.RENAMED
             files = await aria2.torrents_files("gidA")
 
         assert files == [{"name": "new.mkv", "size": 4}]
@@ -921,7 +1032,7 @@ class TestTorrentsRenameFile:
                 "gidA", "old.mkv", os.path.join("Season 01", "new.mkv")
             )
 
-        assert result is True
+        assert result.outcome is RenameOutcome.RENAMED
         assert (tmp_path / "Season 01" / "new.mkv").exists()
 
     async def test_returns_false_when_source_file_missing(self, tmp_path):
@@ -933,7 +1044,95 @@ class TestTorrentsRenameFile:
         with patch.object(aria2, "_call", AsyncMock(side_effect=fake_call)):
             result = await aria2.torrents_rename_file("gidA", "missing.mkv", "new.mkv")
 
-        assert result is False
+        assert result.outcome is RenameOutcome.RETRYABLE_FAILURE
+
+    async def test_recovers_move_completed_before_sidecar_commit(self, tmp_path):
+        aria2 = _aria2()
+        old_file = tmp_path / "old.mkv"
+        new_file = tmp_path / "new.mkv"
+        old_file.write_bytes(b"data")
+        intent = _rename_intent("old.mkv", "new.mkv", old_file)
+        async with Database() as db:
+            await db.aria2.set_rename_intent("gidA", intent)
+        old_file.rename(new_file)
+
+        async def fake_call(method, params=None, timeout=10.0):
+            if method == "getFiles":
+                return [{"path": str(tmp_path / "old.mkv"), "length": "4"}]
+            return {"dir": str(tmp_path)}
+
+        with patch.object(aria2, "_call", AsyncMock(side_effect=fake_call)):
+            result = await aria2.torrents_rename_file("gidA", "old.mkv", "new.mkv")
+            files = await aria2.torrents_files("gidA")
+
+        assert result.outcome is RenameOutcome.ALREADY_APPLIED
+        assert files == [{"name": "new.mkv", "size": 4}]
+        async with Database() as db:
+            record = await db.aria2.get("gidA")
+            assert record is not None
+            assert record.rename_intent is None
+
+    async def test_missing_source_does_not_claim_unrelated_target_without_intent(
+        self, tmp_path
+    ):
+        aria2 = _aria2()
+        target = tmp_path / "new.mkv"
+        target.write_bytes(b"unrelated")
+
+        async def fake_call(method, params=None, timeout=10.0):
+            return {"dir": str(tmp_path)}
+
+        with patch.object(aria2, "_call", AsyncMock(side_effect=fake_call)):
+            result = await aria2.torrents_rename_file("gidA", "old.mkv", "new.mkv")
+
+        assert result.outcome is RenameOutcome.DESTINATION_EXISTS
+        assert target.read_bytes() == b"unrelated"
+        async with Database() as db:
+            assert await db.aria2.get_renamed_paths("gidA") == {}
+
+    async def test_crash_recovery_rejects_target_with_mismatched_fingerprint(
+        self, tmp_path
+    ):
+        aria2 = _aria2()
+        old_file = tmp_path / "old.mkv"
+        old_file.write_bytes(b"original")
+        intent = _rename_intent("old.mkv", "new.mkv", old_file)
+        async with Database() as db:
+            await db.aria2.set_rename_intent("gidA", intent)
+        old_file.unlink()
+        (tmp_path / "new.mkv").write_bytes(b"unrelated")
+
+        async def fake_call(method, params=None, timeout=10.0):
+            return {"dir": str(tmp_path)}
+
+        with patch.object(aria2, "_call", AsyncMock(side_effect=fake_call)):
+            result = await aria2.torrents_rename_file("gidA", "old.mkv", "new.mkv")
+
+        assert result.outcome is RenameOutcome.DESTINATION_EXISTS
+        async with Database() as db:
+            record = await db.aria2.get("gidA")
+            assert record is not None
+            assert record.rename_intent is None
+            assert await db.aria2.get_renamed_paths("gidA") == {}
+
+    async def test_move_failure_clears_rename_intent(self, tmp_path):
+        aria2 = _aria2()
+        (tmp_path / "old.mkv").write_bytes(b"data")
+
+        async def fake_call(method, params=None, timeout=10.0):
+            return {"dir": str(tmp_path)}
+
+        with (
+            patch.object(aria2, "_call", AsyncMock(side_effect=fake_call)),
+            patch.object(aria2, "_move_file", side_effect=OSError("move failed")),
+        ):
+            result = await aria2.torrents_rename_file("gidA", "old.mkv", "new.mkv")
+
+        assert result.outcome is RenameOutcome.RETRYABLE_FAILURE
+        async with Database() as db:
+            record = await db.aria2.get("gidA")
+            assert record is not None
+            assert record.rename_intent is None
 
     async def test_torrents_rename_file_existing_destination_refuses_overwrite(
         self, tmp_path
@@ -949,7 +1148,7 @@ class TestTorrentsRenameFile:
         with patch.object(aria2, "_call", AsyncMock(side_effect=fake_call)):
             result = await aria2.torrents_rename_file("gidA", "old.mkv", "new.mkv")
 
-        assert result is False
+        assert result.outcome is RenameOutcome.DESTINATION_EXISTS
         # 两个文件都原封不动。
         assert (tmp_path / "old.mkv").read_bytes() == b"source data"
         assert (tmp_path / "new.mkv").read_bytes() == b"precious existing data"
@@ -960,7 +1159,7 @@ class TestTorrentsRenameFile:
             aria2, "_call", AsyncMock(side_effect=Aria2ConnectionError("down"))
         ):
             result = await aria2.torrents_rename_file("gidA", "old.mkv", "new.mkv")
-        assert result is False
+        assert result.outcome is RenameOutcome.RETRYABLE_FAILURE
 
 
 # ---------------------------------------------------------------------------
@@ -1015,10 +1214,77 @@ class TestTorrentsDelete:
                 raise Aria2RpcError(1, "Internal error")
             return None
 
+        async with Database() as db:
+            await db.aria2.upsert(
+                "gidA", bangumi_id=1, renamed_paths='{"old.mkv": "staged.mkv"}'
+            )
+
         with patch.object(aria2, "_call", AsyncMock(side_effect=fake_call)):
             result = await aria2.torrents_delete("gidA", delete_files=False)
 
         assert result is False
+        async with Database() as db:
+            record = await db.aria2.get("gidA")
+        assert record is not None
+        assert record.renamed_paths == '{"old.mkv": "staged.mkv"}'
+
+    async def test_delete_uses_current_renamed_path_not_reoccupied_original(
+        self, tmp_path
+    ):
+        aria2 = _aria2()
+        original = tmp_path / "Show S01E01.mkv"
+        staged = tmp_path / ".Show S01E01.ab-old.mkv"
+        staged.write_bytes(b"old-v1")
+        original.write_bytes(b"new-v2")
+
+        async def fake_call(method, params=None, timeout=10.0):
+            if method == "tellStatus":
+                return {"dir": str(tmp_path)}
+            if method == "getFiles":
+                return [{"path": str(original)}]
+            if method == "forceRemove":
+                return "gidA"
+            return None
+
+        async with Database() as db:
+            await db.aria2.upsert("gidA", bangumi_id=1)
+            await db.aria2.set_renamed_path("gidA", original.name, staged.name)
+
+        with patch.object(aria2, "_call", AsyncMock(side_effect=fake_call)):
+            result = await aria2.torrents_delete("gidA", delete_files=True)
+
+        assert result is True
+        assert not staged.exists()
+        assert original.read_bytes() == b"new-v2"
+        async with Database() as db:
+            assert await db.aria2.get("gidA") is None
+
+    async def test_file_listing_failure_preserves_sidecar_and_task(self):
+        aria2 = _aria2()
+        calls = []
+
+        async def fake_call(method, params=None, timeout=10.0):
+            calls.append(method)
+            if method == "tellStatus":
+                return {"dir": "/downloads"}
+            if method == "getFiles":
+                raise Aria2ConnectionError("down")
+            if method == "forceRemove":
+                return "gidA"
+            return None
+
+        async with Database() as db:
+            await db.aria2.upsert(
+                "gidA", bangumi_id=1, renamed_paths='{"old.mkv": "staged.mkv"}'
+            )
+
+        with patch.object(aria2, "_call", AsyncMock(side_effect=fake_call)):
+            result = await aria2.torrents_delete("gidA", delete_files=True)
+
+        assert result is False
+        assert "forceRemove" not in calls
+        async with Database() as db:
+            assert await db.aria2.get("gidA") is not None
 
     async def test_not_found_error_is_treated_as_already_gone(self):
         aria2 = _aria2()
@@ -1037,6 +1303,26 @@ class TestTorrentsDelete:
 
         assert result is True
 
+    async def test_not_found_before_file_listing_clears_stale_sidecar(self):
+        aria2 = _aria2()
+
+        async def fake_call(method, params=None, timeout=10.0):
+            if method == "tellStatus":
+                raise Aria2RpcError(1, "GID#gidA is not found")
+            raise AssertionError(f"unexpected RPC after not-found: {method}")
+
+        async with Database() as db:
+            await db.aria2.upsert(
+                "gidA", bangumi_id=1, renamed_paths='{"old.mkv": "staged.mkv"}'
+            )
+
+        with patch.object(aria2, "_call", AsyncMock(side_effect=fake_call)):
+            result = await aria2.torrents_delete("gidA", delete_files=True)
+
+        assert result is True
+        async with Database() as db:
+            assert await db.aria2.get("gidA") is None
+
     async def test_accepts_pipe_joined_hashes(self):
         aria2 = _aria2()
         seen_gids = []

backend/src/test/test_config.py

@@ -68,6 +68,7 @@ class TestConfigDefaults:
         assert config.bangumi_manage.rename_method == "pn"
         assert config.bangumi_manage.group_tag is False
         assert config.bangumi_manage.remove_bad_torrent is False
+        assert config.bangumi_manage.revision_conflict_policy == "hold"
         assert config.bangumi_manage.eps_complete is False
 
     def test_proxy_defaults(self):
@@ -380,6 +381,21 @@ class TestEnvOverrides:
 
         assert s.rss_parser.engine == "tokenizer"
 
+    def test_revision_conflict_policy_from_env(self, tmp_path):
+        config_file = tmp_path / "config.json"
+
+        with patch.dict(
+            os.environ,
+            {"AB_REVISION_CONFLICT_POLICY": "REPLACE"},
+            clear=False,
+        ):
+            with patch("module.conf.config.CONFIG_PATH", config_file):
+                s = Settings.__new__(Settings)
+                Config.__init__(s)
+                s.init()
+
+        assert s.bangumi_manage.revision_conflict_policy == "replace"
+
 
 # ---------------------------------------------------------------------------
 # Security model
@@ -601,6 +617,9 @@ class TestDefaultSettings:
         """Factory defaults preserve the pre-Preview parser behaviour."""
         assert DEFAULT_SETTINGS["rss_parser"]["engine"] == "classic"
 
+    def test_revision_conflict_policy_defaults_to_hold(self):
+        assert DEFAULT_SETTINGS["bangumi_manage"]["revision_conflict_policy"] == "hold"
+
 
 # ---------------------------------------------------------------------------
 # BCOLORS utility

backend/src/test/test_database_rename_operation.py

@@ -0,0 +1,270 @@
+"""RenameOperationDatabase — #1078 durable rename/replacement state."""
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from sqlalchemy.exc import IntegrityError
+
+from module.database import Database
+from module.database.rename_operation import RenameOperationDatabase
+from module.models import RenameOperation
+
+
+def _operation(**overrides) -> RenameOperation:
+    values = {
+        "downloader_type": "qbittorrent",
+        "kind": "replacement",
+        "state": "planned",
+        "new_task_id": "new-v2",
+        "old_task_id": "old-v1",
+        "save_path": "/downloads/Anime/Season 1",
+        "source_path": "[ANi] Anime - 01 [V2].mp4",
+        "target_path": "Anime S01E01.mp4",
+        "staged_path": ".ab-replace/old-v1/Anime S01E01.mp4",
+        "bangumi_id": 42,
+        "media_type": "episode",
+        "season": 1,
+        "episode": 1.0,
+        "group_name": "ANi",
+        "resolution": "1080p",
+        "old_revision": 1,
+        "new_revision": 2,
+        "revision_metadata": '{"old":"V1","new":"V2"}',
+    }
+    values.update(overrides)
+    return RenameOperation(**values)
+
+
+async def test_get_or_create_persists_complete_operation_metadata(db_session):
+    repo = RenameOperationDatabase(db_session)
+
+    row, created = await repo.get_or_create(_operation())
+
+    assert created is True
+    assert row.id is not None
+    assert row.old_task_id == "old-v1"
+    assert row.staged_path == ".ab-replace/old-v1/Anime S01E01.mp4"
+    assert row.old_revision == 1
+    assert row.new_revision == 2
+    assert row.revision_metadata == '{"old":"V1","new":"V2"}'
+    assert row.attempt_count == 0
+    assert row.created_at is not None
+    assert row.updated_at is not None
+
+    same, duplicate_created = await repo.get_or_create(_operation())
+    assert duplicate_created is False
+    assert same.id == row.id
+    assert await repo.get(row.id) == row
+
+
+async def test_active_target_is_unique_but_done_history_does_not_block_v3(db_session):
+    repo = RenameOperationDatabase(db_session)
+    first, _ = await repo.get_or_create(_operation(state="conflict"))
+
+    with pytest.raises(IntegrityError):
+        await repo.get_or_create(
+            _operation(
+                new_task_id="new-v3",
+                source_path="[ANi] Anime - 01 [V3].mp4",
+                new_revision=3,
+            )
+        )
+
+    await repo.set_state(first.id, "done")
+    v3, created = await repo.get_or_create(
+        _operation(
+            new_task_id="new-v3",
+            source_path="[ANi] Anime - 01 [V3].mp4",
+            new_revision=3,
+        )
+    )
+    assert created is True
+    assert v3.new_revision == 3
+
+
+async def test_upsert_conflict_is_idempotent_and_refreshes_error(db_session):
+    repo = RenameOperationDatabase(db_session)
+
+    first, created = await repo.upsert_conflict(
+        _operation(state="retry", last_error="qB 409")
+    )
+    assert created is True
+    assert first.state == "conflict"
+    assert first.retry_at is None
+
+    same, created_again = await repo.upsert_conflict(
+        _operation(
+            state="planned",
+            old_task_id="resolved-old-owner",
+            last_error="destination still exists",
+        )
+    )
+    assert created_again is False
+    assert same.id == first.id
+    assert same.state == "conflict"
+    assert same.old_task_id == "resolved-old-owner"
+    assert same.last_error == "destination still exists"
+
+
+async def test_claim_is_atomic_due_only_and_increments_attempt(db_session):
+    repo = RenameOperationDatabase(db_session)
+    now = datetime.now(timezone.utc)
+    row, _ = await repo.get_or_create(
+        _operation(state="retry", retry_at=now - timedelta(seconds=1))
+    )
+
+    claimed = await repo.claim(
+        row.id,
+        from_states=("retry",),
+        to_state="planned",
+        now=now,
+    )
+
+    assert claimed is not None
+    assert claimed.state == "planned"
+    assert claimed.attempt_count == 1
+    assert claimed.retry_at is None
+    assert (
+        await repo.claim(
+            row.id,
+            from_states=("retry",),
+            to_state="planned",
+            now=now,
+        )
+        is None
+    )
+
+    future, _ = await repo.get_or_create(
+        _operation(
+            new_task_id="future-v3",
+            source_path="future-v3.mkv",
+            target_path="Anime S01E02.mp4",
+            state="retry",
+            retry_at=now + timedelta(hours=1),
+        )
+    )
+    assert (
+        await repo.claim(
+            future.id,
+            from_states=("retry",),
+            to_state="planned",
+            now=now,
+        )
+        is None
+    )
+
+
+async def test_mark_notified_is_one_shot(db_session):
+    repo = RenameOperationDatabase(db_session)
+    row, _ = await repo.upsert_conflict(_operation())
+    notified_at = datetime.now(timezone.utc)
+
+    assert await repo.mark_notified(row.id, notified_at) is True
+    assert await repo.mark_notified(row.id, notified_at + timedelta(seconds=1)) is False
+
+    loaded = await repo.get(row.id)
+    assert loaded is not None
+    assert loaded.notified_at == notified_at
+
+
+async def test_recover_stale_running_uses_compare_and_swap(db_session):
+    repo = RenameOperationDatabase(db_session)
+    now = datetime.now(timezone.utc)
+    row, _ = await repo.get_or_create(_operation(state="running"))
+    row.updated_at = now - timedelta(minutes=10)
+    db_session.add(row)
+    await db_session.commit()
+
+    assert await repo.recover_stale_running(row.id, before=now - timedelta(minutes=5))
+    assert not await repo.recover_stale_running(
+        row.id, before=now - timedelta(minutes=5)
+    )
+    loaded = await repo.get(row.id)
+    assert loaded is not None
+    assert loaded.state == "retry"
+
+
+async def test_replacement_phase_lease_fences_workers_and_state_commit(db_session):
+    repo = RenameOperationDatabase(db_session)
+    now = datetime.now(timezone.utc)
+    row, _ = await repo.get_or_create(_operation())
+
+    first = await repo.claim_replacement_lease(row.id, owner="worker-a", now=now)
+    second = await repo.claim_replacement_lease(row.id, owner="worker-b", now=now)
+
+    assert first is not None
+    assert first.lease_owner == "worker-a"
+    assert first.attempt_count == 1
+    assert second is None
+    assert (
+        await repo.set_state_claimed(row.id, owner="worker-b", state="old_staged")
+        is None
+    )
+
+    advanced = await repo.set_state_claimed(
+        row.id, owner="worker-a", state="old_staged"
+    )
+    assert advanced is not None
+    assert advanced.state == "old_staged"
+    assert advanced.lease_owner is None
+    assert advanced.lease_expires_at is None
+
+
+async def test_expired_replacement_phase_lease_can_be_reclaimed(db_session):
+    repo = RenameOperationDatabase(db_session)
+    now = datetime.now(timezone.utc)
+    row, _ = await repo.get_or_create(_operation())
+    assert await repo.claim_replacement_lease(
+        row.id,
+        owner="crashed-worker",
+        now=now - timedelta(minutes=10),
+        lease_for=timedelta(minutes=5),
+    )
+
+    reclaimed = await repo.claim_replacement_lease(
+        row.id, owner="recovery-worker", now=now
+    )
+
+    assert reclaimed is not None
+    assert reclaimed.lease_owner == "recovery-worker"
+    assert reclaimed.attempt_count == 2
+
+
+async def test_retry_query_and_done_pruning(db_session):
+    repo = RenameOperationDatabase(db_session)
+    now = datetime.now(timezone.utc)
+    due, _ = await repo.get_or_create(
+        _operation(state="retry", retry_at=now - timedelta(seconds=1))
+    )
+    await repo.get_or_create(
+        _operation(
+            new_task_id="later-v3",
+            source_path="later-v3.mkv",
+            target_path="Anime S01E02.mp4",
+            state="retry",
+            retry_at=now + timedelta(hours=1),
+        )
+    )
+    conflict, _ = await repo.upsert_conflict(
+        _operation(
+            new_task_id="conflict-v4",
+            source_path="conflict-v4.mkv",
+            target_path="Anime S01E03.mp4",
+        )
+    )
+
+    assert [row.id for row in await repo.list_retryable(now)] == [due.id]
+    assert [row.id for row in await repo.list_conflicts()] == [conflict.id]
+
+    await repo.set_state(conflict.id, "done")
+    conflict.updated_at = now - timedelta(days=31)
+    db_session.add(conflict)
+    await db_session.commit()
+    assert await repo.prune_done(before=now - timedelta(days=30)) == 1
+
+
+async def test_database_facade_exposes_rename_operation_repository(db_engine):
+    async with Database(engine=db_engine) as db:
+        row, created = await db.rename_operation.get_or_create(_operation())
+        assert created is True
+        assert await db.rename_operation.get(row.id) is not None

backend/src/test/test_download_client.py

@@ -4,6 +4,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
+from module.downloader.base import RenameOutcome, RenameResult
 from module.downloader.download_client import (
     TORRENT_FETCH_PER_HOST_DELAY,
     AddResult,
@@ -335,27 +336,44 @@ class TestClientDelegation:
         )
         assert len(result) == 1
 
+    async def test_torrent_exists_preserves_unknown_state(
+        self, download_client, mock_qb_client
+    ):
+        mock_qb_client.torrent_exists.return_value = None
+
+        result = await download_client.torrent_exists("hash1")
+
+        assert result is None
+        mock_qb_client.torrent_exists.assert_awaited_once_with("hash1")
+
     async def test_rename_torrent_file_success(self, download_client, mock_qb_client):
-        """rename_torrent_file returns True on success."""
-        mock_qb_client.torrents_rename_file.return_value = True
+        """rename_torrent_file preserves the concrete structured result."""
+        expected = RenameResult(RenameOutcome.RENAMED)
+        mock_qb_client.torrents_rename_file.return_value = expected
         result = await download_client.rename_torrent_file(
             "hash1", "old.mkv", "new.mkv"
         )
-        assert result is True
+        assert result is expected
 
     async def test_rename_torrent_file_failure(self, download_client, mock_qb_client):
-        """rename_torrent_file returns False on failure."""
-        mock_qb_client.torrents_rename_file.return_value = False
+        """Failure reasons are not collapsed back into a boolean."""
+        expected = RenameResult(
+            RenameOutcome.RETRYABLE_FAILURE, detail="verification timed out"
+        )
+        mock_qb_client.torrents_rename_file.return_value = expected
         result = await download_client.rename_torrent_file(
             "hash1", "old.mkv", "new.mkv"
         )
-        assert result is False
+        assert result is expected
+        assert result.outcome is RenameOutcome.RETRYABLE_FAILURE
 
     async def test_rename_torrent_file_passes_verify_flag(
         self, download_client, mock_qb_client
     ):
         """rename_torrent_file forwards the verify kwarg to the underlying client."""
-        mock_qb_client.torrents_rename_file.return_value = True
+        mock_qb_client.torrents_rename_file.return_value = RenameResult(
+            RenameOutcome.RENAMED
+        )
         await download_client.rename_torrent_file(
             "hash1", "old.mkv", "new.mkv", verify=False
         )

backend/src/test/test_downloader_protocol.py

@@ -9,6 +9,8 @@ from module.downloader.base import (
     CoreDownloaderClient,
     DownloaderCapabilities,
     DownloaderClient,
+    RenameOutcome,
+    RenameResult,
 )
 from module.downloader.client.aria2_downloader import Aria2Downloader
 from module.downloader.client.mock_downloader import MockDownloader
@@ -48,6 +50,16 @@ class TestCapabilities:
             assert isinstance(cls.capabilities, DownloaderCapabilities)
 
 
+class TestRenameResult:
+    def test_success_outcomes_preserve_boolean_compatibility(self):
+        assert RenameResult(RenameOutcome.RENAMED)
+        assert RenameResult(RenameOutcome.ALREADY_APPLIED)
+
+    def test_failure_outcomes_are_false(self):
+        assert not RenameResult(RenameOutcome.DESTINATION_EXISTS)
+        assert not RenameResult(RenameOutcome.RETRYABLE_FAILURE)
+
+
 class TestProtocolConformance:
     def test_qb_satisfies_full_protocol(self):
         assert isinstance(_qb(), DownloaderClient)

backend/src/test/test_loops.py

@@ -21,6 +21,7 @@ from module.core.loops import (
 from module.models.bangumi import Notification
 from module.notification.events import (
     OffsetReviewEvent,
+    RenameConflictEvent,
     RssFailureEvent,
     UpdateAvailableEvent,
 )
@@ -144,6 +145,32 @@ class TestRssTick:
 
 
 class TestRenameTick:
+    async def test_persists_and_dispatches_conflict_event_when_push_disabled(self):
+        event = RenameConflictEvent(
+            task_id="new-v2",
+            torrent_name="Episode V2",
+            target_path="Show S01E01.mkv",
+            reason="target exists",
+        )
+        notifier = AsyncMock()
+        mock_renamer = AsyncMock()
+        mock_renamer.rename = AsyncMock(return_value=[])
+        mock_renamer.events = [event]
+
+        with (
+            patch(
+                "module.core.loops.DownloadClient",
+                return_value=_async_cm(AsyncMock()),
+            ),
+            patch("module.core.loops.Renamer", return_value=mock_renamer),
+            patch("module.core.loops.settings") as mock_settings,
+        ):
+            mock_settings.notification.enable = False
+            await rename_tick(notifier)
+
+        notifier.send_event.assert_awaited_once_with(event)
+        notifier.send_all.assert_not_awaited()
+
     async def test_sends_notification_per_renamed_item(self):
         """Every item Renamer.rename() returns is forwarded to the notifier."""
         notify = Notification(official_title="Test Anime", season=1, episode=1)

backend/src/test/test_migrations_module.py

@@ -375,6 +375,14 @@ class TestRunMigrations:
         aria2_cols = {c["name"] for c in inspector.get_columns("aria2_gid")}
         assert "renamed_paths" in aria2_cols
 
+    def test_v24_adds_aria2_rename_intent_column(self):
+        """v24 persists an ownership proof before moving an aria2 file."""
+        engine = _make_v0_engine()
+        run_migrations(engine)
+
+        aria2_cols = _columns(engine, "aria2_gid")
+        assert "rename_intent" in aria2_cols
+
     def test_creates_inboxmessage_table(self):
         """v16 creates the in-app notification center table with its indexes."""
         engine = _make_v0_engine()
@@ -399,6 +407,110 @@ class TestRunMigrations:
         indexes = {ix["name"]: ix for ix in inspector.get_indexes("llmcredential")}
         assert indexes["ix_llmcredential_provider_id"]["unique"]
 
+    def test_v23_creates_durable_rename_operation_schema(self):
+        engine = _make_v0_engine()
+
+        run_migrations(engine)
+
+        inspector = inspect(engine)
+        assert "rename_operation" in inspector.get_table_names()
+        assert {
+            "downloader_type",
+            "kind",
+            "state",
+            "new_task_id",
+            "old_task_id",
+            "save_path",
+            "source_path",
+            "target_path",
+            "staged_path",
+            "bangumi_id",
+            "media_type",
+            "season",
+            "episode",
+            "group_name",
+            "resolution",
+            "old_revision",
+            "new_revision",
+            "revision_metadata",
+            "attempt_count",
+            "retry_at",
+            "notified_at",
+            "last_error",
+            "created_at",
+            "updated_at",
+        } <= _columns(engine, "rename_operation")
+        indexes = {
+            item["name"]: item for item in inspector.get_indexes("rename_operation")
+        }
+        assert {
+            "ux_rename_operation_identity",
+            "ux_rename_operation_active_target",
+            "ix_rename_operation_state_retry_at",
+            "ix_rename_operation_new_task_id",
+            "ix_rename_operation_old_task_id",
+        } <= set(indexes)
+        assert indexes["ux_rename_operation_identity"]["unique"]
+        assert indexes["ux_rename_operation_active_target"]["unique"]
+
+    def test_v23_active_target_unique_index_ignores_done_history(self):
+        engine = _make_v0_engine()
+        run_migrations(engine)
+        insert = text(
+            "INSERT INTO rename_operation "
+            "(downloader_type, kind, state, new_task_id, save_path, "
+            " source_path, target_path, attempt_count, created_at, updated_at) "
+            "VALUES ('qbittorrent', 'replacement', :state, :task, '/show', "
+            " :source, 'Show S01E01.mkv', 0, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
+        )
+        with engine.begin() as conn:
+            conn.execute(insert, {"state": "conflict", "task": "v2", "source": "v2"})
+        with pytest.raises(IntegrityError):
+            with engine.begin() as conn:
+                conn.execute(
+                    insert,
+                    {"state": "planned", "task": "v3", "source": "v3"},
+                )
+        with engine.begin() as conn:
+            conn.execute(
+                text(
+                    "UPDATE rename_operation SET state = 'done' WHERE new_task_id = 'v2'"
+                )
+            )
+            conn.execute(
+                insert,
+                {"state": "planned", "task": "v3", "source": "v3"},
+            )
+
+    def test_v23_rejects_unknown_operation_state(self):
+        engine = _make_v0_engine()
+        run_migrations(engine)
+
+        with pytest.raises(IntegrityError):
+            with engine.begin() as conn:
+                conn.execute(
+                    text(
+                        "INSERT INTO rename_operation "
+                        "(downloader_type, kind, state, new_task_id, save_path, "
+                        " source_path, target_path, attempt_count, created_at, updated_at) "
+                        "VALUES ('qbittorrent', 'conflict', 'unknown', 'v2', '/show', "
+                        " 'v2.mkv', 'Show S01E01.mkv', 0, CURRENT_TIMESTAMP, "
+                        " CURRENT_TIMESTAMP)"
+                    )
+                )
+
+    def test_v23_guard_accepts_metadata_created_table(self, monkeypatch):
+        engine = _make_v0_engine()
+        _run_through_version(engine, 22, monkeypatch)
+        SQLModel.metadata.create_all(engine)
+
+        run_migrations(engine)
+
+        with engine.connect() as conn:
+            assert get_schema_version(conn) == CURRENT_SCHEMA_VERSION
+        indexes = {ix["name"] for ix in inspect(engine).get_indexes("rename_operation")}
+        assert "ux_rename_operation_active_target" in indexes
+
     def test_upgrades_auth_beta_v20_without_losing_tokens(self, monkeypatch):
         engine = _make_v19_auth_engine()
         _run_through_version(engine, 20, monkeypatch)

backend/src/test/test_mock_downloader.py

@@ -2,7 +2,7 @@
 
 import pytest
 
-from module.downloader import AddResult
+from module.downloader import AddResult, RenameOutcome
 from module.downloader.client.mock_downloader import MockDownloader
 
 
@@ -192,6 +192,12 @@ class TestMockDownloaderTorrentsInfo:
         result = await mock_dl.torrents_info(status_filter=None, category="Bangumi")
         assert result == []
 
+    async def test_torrent_exists_distinguishes_present_and_absent(self, mock_dl):
+        torrent_hash = mock_dl.add_mock_torrent("Anime A", category="Bangumi")
+
+        assert await mock_dl.torrent_exists(torrent_hash) is True
+        assert await mock_dl.torrent_exists("missing") is False
+
 
 class TestMockDownloaderTorrentsFiles:
     async def test_returns_files_for_known_hash(self, mock_dl):
@@ -257,7 +263,7 @@ class TestMockDownloaderRename:
             old_path="old.mkv",
             new_path="new.mkv",
         )
-        assert result is True
+        assert result.outcome is RenameOutcome.RENAMED
 
     async def test_rename_with_verify_flag(self, mock_dl):
         result = await mock_dl.torrents_rename_file(
@@ -266,7 +272,7 @@ class TestMockDownloaderRename:
             new_path="new.mkv",
             verify=False,
         )
-        assert result is True
+        assert result.outcome is RenameOutcome.RENAMED
 
 
 # ---------------------------------------------------------------------------

backend/src/test/test_qb_downloader.py

@@ -12,7 +12,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import httpx
 import pytest
 
-from module.downloader import AddResult
+from module.downloader import AddResult, RenameOutcome
 from module.downloader.client.qb_downloader import QbDownloader
 
 # ---------------------------------------------------------------------------
@@ -1181,6 +1181,34 @@ class TestTorrentsInfoPausedFilter:
         assert sent_params["filter"] == "completed"
 
 
+class TestTorrentExists:
+    async def test_queries_exact_hash_and_confirms_presence(self):
+        qb = QbDownloader(host="localhost:8080", username="u", password="p", ssl=False)
+        qb._client = AsyncMock()
+        response = MagicMock(status_code=200)
+        response.json.return_value = [{"hash": "ABC123"}]
+        qb._client.request = AsyncMock(return_value=response)
+
+        assert await qb.torrent_exists("abc123") is True
+        assert qb._client.request.call_args.kwargs["params"] == {"hashes": "abc123"}
+
+    async def test_empty_exact_hash_query_confirms_absence(self):
+        qb = QbDownloader(host="localhost:8080", username="u", password="p", ssl=False)
+        qb._client = AsyncMock()
+        response = MagicMock(status_code=200)
+        response.json.return_value = []
+        qb._client.request = AsyncMock(return_value=response)
+
+        assert await qb.torrent_exists("missing") is False
+
+    async def test_failed_query_is_unknown_not_absent(self):
+        qb = QbDownloader(host="localhost:8080", username="u", password="p", ssl=False)
+        qb._client = AsyncMock()
+        qb._client.request = AsyncMock(side_effect=ConnectionError("offline"))
+
+        assert await qb.torrent_exists("abc123") is None
+
+
 # ---------------------------------------------------------------------------
 # torrents_delete (#1046)
 # ---------------------------------------------------------------------------
@@ -1250,26 +1278,42 @@ class TestRenameVerifyContract:
         return qb
 
     async def test_verify_false_when_file_keeps_old_name(self):
-        """API 200 but the file never gets renamed -> False."""
+        """API 200 but the file never gets renamed -> retryable."""
         qb = self._make_qb([{"name": "old.mkv"}])
-        assert await qb.torrents_rename_file("h", "old.mkv", "new.mkv") is False
+        result = await qb.torrents_rename_file("h", "old.mkv", "new.mkv")
+        assert result.outcome is RenameOutcome.RETRYABLE_FAILURE
 
     async def test_verify_false_when_neither_name_present(self):
-        """Neither old nor new name in the file list -> False, not a blind True."""
+        """Neither old nor new name is retryable, not a blind success."""
         qb = self._make_qb([{"name": "unrelated.mkv"}])
-        assert await qb.torrents_rename_file("h", "old.mkv", "new.mkv") is False
+        result = await qb.torrents_rename_file("h", "old.mkv", "new.mkv")
+        assert result.outcome is RenameOutcome.RETRYABLE_FAILURE
 
     async def test_verify_true_when_new_path_appears(self):
         """A real, verified rename returns True."""
         qb = self._make_qb([{"name": "new.mkv"}])
-        assert await qb.torrents_rename_file("h", "old.mkv", "new.mkv") is True
+        result = await qb.torrents_rename_file("h", "old.mkv", "new.mkv")
+        assert result.outcome is RenameOutcome.RENAMED
 
     async def test_verify_409_conflict_is_false(self):
-        """Target already exists (duplicate from another source) -> False."""
+        """Target already exists is a terminal, distinguishable outcome."""
         qb = self._make_qb([{"name": "old.mkv"}], post_code=409)
-        assert await qb.torrents_rename_file("h", "old.mkv", "new.mkv") is False
+        result = await qb.torrents_rename_file("h", "old.mkv", "new.mkv")
+        assert result.outcome is RenameOutcome.DESTINATION_EXISTS
 
     async def test_rename_204_with_verified_new_path_is_true(self):
         """qB 5.2 对成功的 renameFile 回 204；不能再用 ==200 判定失败。"""
         qb = self._make_qb([{"name": "new.mkv"}], post_code=204)
-        assert await qb.torrents_rename_file("h", "old.mkv", "new.mkv") is True
+        result = await qb.torrents_rename_file("h", "old.mkv", "new.mkv")
+        assert result.outcome is RenameOutcome.RENAMED
+
+    async def test_http_error_is_retryable(self):
+        qb = self._make_qb([{"name": "old.mkv"}], post_code=500)
+        result = await qb.torrents_rename_file("h", "old.mkv", "new.mkv")
+        assert result.outcome is RenameOutcome.RETRYABLE_FAILURE
+
+    async def test_network_error_is_retryable(self):
+        qb = self._make_qb([])
+        qb._post = AsyncMock(side_effect=httpx.ConnectError("down"))
+        result = await qb.torrents_rename_file("h", "old.mkv", "new.mkv")
+        assert result.outcome is RenameOutcome.RETRYABLE_FAILURE

backend/src/test/test_renamer.py

@@ -5,8 +5,8 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 
 from module.conf import settings
-from module.downloader import DownloadClient
-from module.manager.renamer import Renamer
+from module.downloader import DownloadClient, RenameOutcome, RenameResult
+from module.manager.renamer import PreparedMediaRename, Renamer
 from module.models import EpisodeFile, Notification, SubtitleFile
 
 # ---------------------------------------------------------------------------
@@ -996,6 +996,519 @@ class TestRenameFlow:
         renamer.client.client.torrents_rename_file.assert_not_called()
 
 
+class TestRevisionConflictFlow:
+    V1 = "[ANi] 尼古喵喵 - 01 [1080P][Baha][WEB-DL][AAC AVC][CHT].mp4"
+    V2 = "[ANi] 尼古喵喵 - 01 [V2][1080P][Baha][WEB-DL][AAC AVC][CHT].mp4"
+    TARGET = "尼古喵喵 S01E01.mp4"
+    SAVE_PATH = "/downloads/Bangumi/尼古喵喵 (2026)/Season 1"
+
+    @pytest.fixture
+    def renamer(self, mock_qb_client):
+        with patch("module.downloader.download_client.settings") as mock_settings:
+            mock_settings.downloader.type = "qbittorrent"
+            mock_settings.downloader.host = "localhost:8080"
+            mock_settings.downloader.username = "admin"
+            mock_settings.downloader.password = "admin"
+            mock_settings.downloader.ssl = False
+            mock_settings.downloader.path = "/downloads/Bangumi"
+            mock_settings.bangumi_manage.group_tag = False
+            mock_settings.bangumi_manage.remove_bad_torrent = False
+            with patch(
+                "module.downloader.download_client.DownloadClient._DownloadClient__getClient",
+                return_value=mock_qb_client,
+            ):
+                client = DownloadClient()
+        client.client = mock_qb_client
+
+        async def torrent_exists(torrent_hash):
+            infos = mock_qb_client.torrents_info.return_value
+            return isinstance(infos, list) and any(
+                item.get("hash") == torrent_hash for item in infos
+            )
+
+        mock_qb_client.torrent_exists.side_effect = torrent_exists
+        return Renamer(client)
+
+    def _infos(self):
+        return [
+            {
+                "hash": "old-v1",
+                "name": self.V1,
+                "save_path": self.SAVE_PATH,
+                "tags": "ab:42, ab:renamed",
+            },
+            {
+                "hash": "new-v2",
+                "name": self.V2,
+                "save_path": self.SAVE_PATH,
+                "tags": "ab:42",
+            },
+        ]
+
+    @staticmethod
+    def _offsets():
+        return {"new-v2": (0, 0, "episode")}
+
+    async def test_renamed_tag_skips_before_file_and_offset_queries(self, renamer):
+        renamer.client.client.torrents_info.return_value = [self._infos()[0]]
+        with patch.object(renamer, "_batch_lookup_offsets", AsyncMock()) as lookup:
+            assert await renamer.rename() == []
+
+        renamer.client.client.torrents_files.assert_not_awaited()
+        lookup.assert_not_awaited()
+
+    async def test_overlapping_ordinary_rename_is_claimed_once(
+        self, renamer, test_settings
+    ):
+        import asyncio
+
+        info = self._infos()[1]
+        renamer.client.client.torrents_info.return_value = [info]
+        renamer.client.client.torrents_files.return_value = [{"name": self.V2}]
+        renamer.client.client.torrents_rename_file.return_value = RenameResult(
+            RenameOutcome.RENAMED
+        )
+        test_settings.bangumi_manage.rename_method = "pn"
+        other = Renamer(renamer.client)
+
+        with (
+            patch("module.manager.renamer.settings", test_settings),
+            patch.object(
+                renamer,
+                "_batch_lookup_offsets",
+                AsyncMock(return_value=self._offsets()),
+            ),
+            patch.object(
+                other,
+                "_batch_lookup_offsets",
+                AsyncMock(return_value=self._offsets()),
+            ),
+        ):
+            results = await asyncio.gather(renamer.rename(), other.rename())
+
+        assert sum(len(result) for result in results) == 1
+        renamer.client.client.torrents_rename_file.assert_awaited_once()
+
+        from module.database import Database
+
+        async with Database() as db:
+            operation = await db.rename_operation.get_by_target(
+                downloader_type=renamer._downloader_type(),
+                save_path=self.SAVE_PATH,
+                target_path=self.TARGET,
+                active_only=False,
+            )
+        assert operation is not None
+        assert operation.state == "done"
+        assert operation.attempt_count == 1
+
+    async def test_ordinary_rename_reconciles_downloader_success_after_crash(
+        self, renamer
+    ):
+        info = self._infos()[1]
+        # qB already exposes the canonical name, but the durable operation row
+        # was not marked done before the previous process stopped.
+        renamer.client.client.torrents_files.return_value = [{"name": self.TARGET}]
+        prepared = PreparedMediaRename(
+            episode=EpisodeFile(
+                media_path=self.V2,
+                title="尼古喵喵",
+                season=1,
+                episode=1,
+                suffix=".mp4",
+            ),
+            source_path=self.V2,
+            target_path=self.TARGET,
+        )
+
+        report = await renamer._run_ordinary_rename(
+            info=info,
+            prepared=prepared,
+            identity=None,
+            bangumi_name="尼古喵喵",
+            episode_offset=0,
+        )
+
+        assert report.result.outcome is RenameOutcome.ALREADY_APPLIED
+        renamer.client.client.torrents_rename_file.assert_not_awaited()
+        from module.database import Database
+
+        async with Database() as db:
+            operation = await db.rename_operation.get_by_target(
+                downloader_type=renamer._downloader_type(),
+                save_path=self.SAVE_PATH,
+                target_path=self.TARGET,
+                active_only=False,
+            )
+        assert operation is not None
+        assert operation.state == "done"
+
+    async def test_hold_conflict_is_persistent_and_notified_once(
+        self, renamer, test_settings
+    ):
+        renamer.client.client.torrents_info.return_value = self._infos()
+
+        async def files(torrent_hash):
+            if torrent_hash == "old-v1":
+                return [{"name": self.TARGET}]
+            return [{"name": self.V2}]
+
+        renamer.client.client.torrents_files.side_effect = files
+        test_settings.bangumi_manage.rename_method = "pn"
+        test_settings.bangumi_manage.revision_conflict_policy = "hold"
+
+        with (
+            patch("module.manager.renamer.settings", test_settings),
+            patch.object(
+                renamer,
+                "_batch_lookup_offsets",
+                AsyncMock(return_value=self._offsets()),
+            ),
+        ):
+            assert await renamer.rename() == []
+
+        assert len(renamer.events) == 1
+        renamer.client.client.torrents_rename_file.assert_not_awaited()
+
+        restarted = Renamer(renamer.client)
+        with (
+            patch("module.manager.renamer.settings", test_settings),
+            patch.object(
+                restarted,
+                "_batch_lookup_offsets",
+                AsyncMock(return_value=self._offsets()),
+            ),
+        ):
+            assert await restarted.rename() == []
+
+        assert restarted.events == []
+        renamer.client.client.torrents_rename_file.assert_not_awaited()
+
+    async def test_replace_stages_promotes_then_deletes_old(
+        self, renamer, test_settings
+    ):
+        infos = self._infos()
+        renamer.client.client.torrents_info.return_value = infos
+        paths = {"old-v1": self.TARGET, "new-v2": self.V2}
+
+        async def files(torrent_hash):
+            return [{"name": paths[torrent_hash]}]
+
+        async def rename(torrent_hash, old_path, new_path, verify=True):
+            assert paths[torrent_hash] == old_path
+            paths[torrent_hash] = new_path
+            return RenameResult(RenameOutcome.RENAMED)
+
+        async def delete(torrent_hash, delete_files=True):
+            infos[:] = [info for info in infos if info["hash"] != torrent_hash]
+            return True
+
+        renamer.client.client.torrents_files.side_effect = files
+        renamer.client.client.torrents_rename_file.side_effect = rename
+        renamer.client.client.torrents_delete.side_effect = delete
+        test_settings.bangumi_manage.rename_method = "pn"
+        test_settings.bangumi_manage.revision_conflict_policy = "replace"
+
+        with (
+            patch("module.manager.renamer.settings", test_settings),
+            patch.object(
+                renamer,
+                "_batch_lookup_offsets",
+                AsyncMock(return_value=self._offsets()),
+            ),
+        ):
+            result = await renamer.rename()
+
+        assert len(result) == 1
+        assert paths["new-v2"] == self.TARGET
+        assert ".ab-replaced-v1-old-v1" in paths["old-v1"]
+        renamer.client.client.torrents_delete.assert_awaited_once_with(
+            "old-v1", delete_files=True
+        )
+        assert renamer.events == []
+
+    async def test_promotion_failure_restores_v1_and_waits_for_retry(
+        self, renamer, test_settings
+    ):
+        renamer.client.client.torrents_info.return_value = self._infos()
+        paths = {"old-v1": self.TARGET, "new-v2": self.V2}
+
+        async def files(torrent_hash):
+            return [{"name": paths[torrent_hash]}]
+
+        async def rename(torrent_hash, old_path, new_path, verify=True):
+            if torrent_hash == "new-v2":
+                return RenameResult(
+                    RenameOutcome.RETRYABLE_FAILURE, detail="qB unavailable"
+                )
+            assert paths[torrent_hash] == old_path
+            paths[torrent_hash] = new_path
+            return RenameResult(RenameOutcome.RENAMED)
+
+        renamer.client.client.torrents_files.side_effect = files
+        renamer.client.client.torrents_rename_file.side_effect = rename
+        renamer.client.client.torrents_delete.return_value = True
+        test_settings.bangumi_manage.rename_method = "pn"
+        test_settings.bangumi_manage.revision_conflict_policy = "replace"
+
+        with (
+            patch("module.manager.renamer.settings", test_settings),
+            patch.object(
+                renamer,
+                "_batch_lookup_offsets",
+                AsyncMock(return_value=self._offsets()),
+            ),
+        ):
+            assert await renamer.rename() == []
+
+        assert paths["old-v1"] == self.TARGET
+        assert paths["new-v2"] == self.V2
+        assert renamer.client.client.torrents_rename_file.await_count == 3
+        renamer.client.client.torrents_delete.assert_not_awaited()
+
+        restarted = Renamer(renamer.client)
+        with (
+            patch("module.manager.renamer.settings", test_settings),
+            patch.object(
+                restarted,
+                "_batch_lookup_offsets",
+                AsyncMock(return_value=self._offsets()),
+            ),
+        ):
+            assert await restarted.rename() == []
+        assert renamer.client.client.torrents_rename_file.await_count == 3
+
+    async def test_multifile_candidate_falls_back_to_hold(self, renamer, test_settings):
+        renamer.client.client.torrents_info.return_value = self._infos()
+
+        async def files(torrent_hash):
+            if torrent_hash == "old-v1":
+                return [{"name": self.TARGET}]
+            return [{"name": self.V2}, {"name": "尼古喵喵 - 01.ass"}]
+
+        renamer.client.client.torrents_files.side_effect = files
+        test_settings.bangumi_manage.rename_method = "pn"
+        test_settings.bangumi_manage.revision_conflict_policy = "replace"
+
+        with (
+            patch("module.manager.renamer.settings", test_settings),
+            patch.object(
+                renamer,
+                "_batch_lookup_offsets",
+                AsyncMock(return_value=self._offsets()),
+            ),
+        ):
+            assert await renamer.rename() == []
+
+        assert len(renamer.events) == 1
+        assert "single-file" in renamer.events[0].reason
+        renamer.client.client.torrents_rename_file.assert_not_awaited()
+        renamer.client.client.torrents_delete.assert_not_awaited()
+
+    async def test_missing_v2_after_staging_restores_v1_before_conflict(
+        self, renamer, test_settings
+    ):
+        renamer.client.client.torrents_info.return_value = self._infos()
+        paths = {"old-v1": self.TARGET, "new-v2": self.V2}
+        new_queries = 0
+
+        async def files(torrent_hash):
+            nonlocal new_queries
+            if torrent_hash == "new-v2":
+                new_queries += 1
+                if new_queries > 1:
+                    return []
+            return [{"name": paths[torrent_hash]}]
+
+        async def rename(torrent_hash, old_path, new_path, verify=True):
+            assert paths[torrent_hash] == old_path
+            paths[torrent_hash] = new_path
+            return RenameResult(RenameOutcome.RENAMED)
+
+        renamer.client.client.torrents_files.side_effect = files
+        renamer.client.client.torrents_rename_file.side_effect = rename
+        test_settings.bangumi_manage.rename_method = "pn"
+        test_settings.bangumi_manage.revision_conflict_policy = "replace"
+
+        with (
+            patch("module.manager.renamer.settings", test_settings),
+            patch.object(
+                renamer,
+                "_batch_lookup_offsets",
+                AsyncMock(return_value=self._offsets()),
+            ),
+        ):
+            assert await renamer.rename() == []
+
+        assert paths["old-v1"] == self.TARGET
+        assert len(renamer.events) == 1
+        assert "V1 was restored" in renamer.events[0].reason
+        renamer.client.client.torrents_delete.assert_not_awaited()
+
+    async def test_multifile_owner_blocks_destructive_replacement(
+        self, renamer, test_settings
+    ):
+        renamer.client.client.torrents_info.return_value = self._infos()
+
+        async def files(torrent_hash):
+            if torrent_hash == "old-v1":
+                return [{"name": self.TARGET}, {"name": "old-release.nfo"}]
+            return [{"name": self.V2}]
+
+        renamer.client.client.torrents_files.side_effect = files
+        test_settings.bangumi_manage.rename_method = "pn"
+        test_settings.bangumi_manage.revision_conflict_policy = "replace"
+
+        with (
+            patch("module.manager.renamer.settings", test_settings),
+            patch.object(
+                renamer,
+                "_batch_lookup_offsets",
+                AsyncMock(return_value=self._offsets()),
+            ),
+        ):
+            assert await renamer.rename() == []
+
+        assert len(renamer.events) == 1
+        assert "single-file" in renamer.events[0].reason
+        renamer.client.client.torrents_rename_file.assert_not_awaited()
+        renamer.client.client.torrents_delete.assert_not_awaited()
+
+    async def test_multi_episode_collection_conflicts_are_persistent(
+        self, renamer, test_settings
+    ):
+        info = {
+            "hash": "season-pack",
+            "name": "[ANi] 尼古喵喵 01-02 [1080P].torrent",
+            "save_path": self.SAVE_PATH,
+            "tags": "ab:42",
+        }
+        renamer.client.client.torrents_info.return_value = [info]
+        renamer.client.client.torrents_files.return_value = [
+            {"name": "raw01.mp4"},
+            {"name": "raw02.mp4"},
+        ]
+        renamer.client.client.torrents_rename_file.return_value = RenameResult(
+            RenameOutcome.DESTINATION_EXISTS, detail="target exists"
+        )
+        test_settings.bangumi_manage.rename_method = "pn"
+        test_settings.bangumi_manage.remove_bad_torrent = True
+        test_settings.bangumi_manage.revision_conflict_policy = "replace"
+
+        def parse(torrent_path, **kwargs):
+            episode = 1 if "01" in torrent_path else 2
+            return EpisodeFile(
+                media_path=torrent_path,
+                title="尼古喵喵",
+                season=1,
+                episode=episode,
+                suffix=".mp4",
+            )
+
+        with (
+            patch("module.manager.renamer.settings", test_settings),
+            patch.object(renamer._parser, "torrent_parser", side_effect=parse),
+            patch.object(
+                renamer,
+                "_batch_lookup_offsets",
+                AsyncMock(return_value={"season-pack": (0, 0, "episode")}),
+            ),
+        ):
+            assert await renamer.rename() == []
+
+        assert renamer.client.client.torrents_rename_file.await_count == 2
+        assert len(renamer.events) == 2
+        renamer.client.client.torrents_delete.assert_not_awaited()
+        renamer.client.client.set_category.assert_not_awaited()
+
+        restarted = Renamer(renamer.client)
+        with (
+            patch("module.manager.renamer.settings", test_settings),
+            patch.object(restarted._parser, "torrent_parser", side_effect=parse),
+            patch.object(
+                restarted,
+                "_batch_lookup_offsets",
+                AsyncMock(return_value={"season-pack": (0, 0, "episode")}),
+            ),
+        ):
+            assert await restarted.rename() == []
+
+        assert renamer.client.client.torrents_rename_file.await_count == 2
+        assert restarted.events == []
+
+    async def test_restart_finishes_forward_cleanup_after_promotion(
+        self, renamer, test_settings
+    ):
+        infos = self._infos()
+        renamer.client.client.torrents_info.return_value = infos
+        paths = {"old-v1": self.TARGET, "new-v2": self.V2}
+
+        async def files(torrent_hash):
+            return [{"name": paths[torrent_hash]}]
+
+        async def rename(torrent_hash, old_path, new_path, verify=True):
+            assert paths[torrent_hash] == old_path
+            paths[torrent_hash] = new_path
+            return RenameResult(RenameOutcome.RENAMED)
+
+        renamer.client.client.torrents_files.side_effect = files
+        renamer.client.client.torrents_rename_file.side_effect = rename
+        renamer.client.client.torrents_delete.return_value = False
+        test_settings.bangumi_manage.rename_method = "pn"
+        test_settings.bangumi_manage.revision_conflict_policy = "replace"
+
+        with (
+            patch("module.manager.renamer.settings", test_settings),
+            patch.object(
+                renamer,
+                "_batch_lookup_offsets",
+                AsyncMock(return_value=self._offsets()),
+            ),
+        ):
+            assert await renamer.rename() == []
+
+        assert paths["new-v2"] == self.TARGET
+        assert ".ab-replaced-v1-old-v1" in paths["old-v1"]
+
+        from datetime import datetime, timedelta, timezone
+
+        from module.database import Database
+
+        async with Database() as db:
+            operation = await db.rename_operation.get_by_target(
+                downloader_type=renamer._downloader_type(),
+                save_path=self.SAVE_PATH,
+                target_path=self.TARGET,
+            )
+            assert operation is not None
+            assert operation.state == "new_promoted"
+            await db.rename_operation.set_state(
+                operation.id,
+                "new_promoted",
+                retry_at=datetime.now(timezone.utc) - timedelta(seconds=1),
+            )
+
+        async def delete(torrent_hash, delete_files=True):
+            infos[:] = [info for info in infos if info["hash"] != torrent_hash]
+            return True
+
+        renamer.client.client.torrents_delete.side_effect = delete
+        restarted = Renamer(renamer.client)
+        with (
+            patch("module.manager.renamer.settings", test_settings),
+            patch.object(
+                restarted,
+                "_batch_lookup_offsets",
+                AsyncMock(return_value=self._offsets()),
+            ),
+        ):
+            result = await restarted.rename()
+
+        assert len(result) == 1
+        assert paths["new-v2"] == self.TARGET
+        assert renamer.client.client.torrents_delete.await_count == 2
+
+
 # ---------------------------------------------------------------------------
 # _parse_bangumi_id_from_tags
 # ---------------------------------------------------------------------------

backend/src/test/test_revision_policy.py

@@ -0,0 +1,58 @@
+from module.manager.revision_policy import (
+    is_strict_upgrade,
+    parse_revision_identity,
+    replacement_staged_path,
+    same_release_identity,
+)
+
+V1 = "[ANi] 尼古喵喵 - 01 [1080P][Baha][WEB-DL][AAC AVC][CHT].mp4"
+V2 = "[ANi] 尼古喵喵 - 01 [V2][1080P][Baha][WEB-DL][AAC AVC][CHT].mp4"
+
+
+def test_mikan_classic_v2_is_a_strict_upgrade():
+    old = parse_revision_identity(V1, bangumi_id=42, default_season=1)
+    new = parse_revision_identity(V2, bangumi_id=42, default_season=1)
+
+    assert old is not None
+    assert new is not None
+    assert old.revision == 1
+    assert new.revision == 2
+    assert same_release_identity(old, new)
+    assert is_strict_upgrade(old, new)
+
+
+def test_cross_group_or_resolution_is_not_a_strict_upgrade():
+    old = parse_revision_identity(V1, bangumi_id=42, default_season=1)
+    other_group = parse_revision_identity(
+        V2.replace("[ANi]", "[Other]"), bangumi_id=42, default_season=1
+    )
+    other_resolution = parse_revision_identity(
+        V2.replace("1080P", "720P"), bangumi_id=42, default_season=1
+    )
+
+    assert old is not None
+    assert other_group is not None
+    assert other_resolution is not None
+    assert not is_strict_upgrade(old, other_group)
+    assert not is_strict_upgrade(old, other_resolution)
+
+
+def test_missing_bangumi_group_or_resolution_is_ineligible():
+    assert parse_revision_identity(V2, bangumi_id=None, default_season=1) is None
+    assert (
+        parse_revision_identity(
+            "尼古喵喵 - 01 [V2][1080P].mp4",
+            bangumi_id=42,
+            default_season=1,
+        )
+        is None
+    )
+
+
+def test_staged_path_is_deterministic_and_keeps_extension():
+    assert (
+        replacement_staged_path(
+            "subdir/尼古喵喵 S01E01.mp4", old_task_id="abcdef123456", old_revision=1
+        )
+        == "subdir/尼古喵喵 S01E01.ab-replaced-v1-abcdef12.mp4"
+    )

docs/plans/2026-07-13-issue-1078-revision-replacement-design.md

@@ -51,6 +51,11 @@ source path, and target path. It stores operation kind, phase, old owner task,
 deterministic temporary path, parsed identity metadata, attempt counters,
 retry time, notification time, last error, and timestamps.
 
+Each replacement phase is fenced by an expiring database lease and committed
+with compare-and-swap semantics. This supplements the target uniqueness index:
+multiple processes may observe the same row, but only the lease holder can run
+the phase and persist its transition.
+
 Normal conflict phases are `CONFLICT` and `RETRY_WAIT`. Replacement phases are:
 
 ```text
@@ -96,11 +101,14 @@ verification after every action. `deleteFiles=true` is allowed only after the
 single-file guard has passed.
 
 For aria2, rename remains an AutoBangumi filesystem move recorded in
-`aria2_gid.renamed_paths`. Old-version cleanup must resolve the old gid's current
-renamed path, delete that deterministic temporary file, call `forceRemove`
-without generic file deletion, and delete the sidecar only after success or an
-explicit not-found response. This prevents cleanup from deleting the V2 file now
-occupying aria2's original path.
+`aria2_gid.renamed_paths`. Before moving a file, aria2 persists an intent with
+the source and target paths plus the source filesystem identity. A restart may
+finalize the move only when the target still has that exact identity; an
+unrelated existing target remains a conflict. Old-version cleanup must resolve
+the old gid's current renamed path, delete that deterministic temporary file,
+call `forceRemove` without generic file deletion, and delete the sidecar only
+after success or an explicit not-found response. This prevents cleanup from
+deleting the V2 file now occupying aria2's original path.
 
 ## Configuration and UI
 

webui/src/components/layout/ab-notification-center.vue

@@ -17,6 +17,7 @@ const KIND_ROUTES: Record<string, string> = {
   rss_failure: '/rss',
   offset_review: '/bangumi',
   download_failure: '/bangumi',
+  rename_conflict: '/downloader',
 };
 
 const SEVERITY_ICONS = {

webui/src/components/setting/__tests__/config-manage.test.ts

@@ -0,0 +1,83 @@
+import { mount } from '@vue/test-utils';
+import { defineComponent, nextTick } from 'vue';
+import ConfigManage from '../config-manage.vue';
+
+vi.mock('@/hooks/useMyI18n', () => ({
+  useMyI18n: () => ({ t: (key: string) => key }),
+}));
+
+vi.mock('@/store/config', async () => {
+  const { computed } = await vi.importActual<typeof import('vue')>('vue');
+  const manageState = {
+    enable: true,
+    eps_complete: false,
+    rename_method: 'pn',
+    revision_conflict_policy: 'hold',
+    group_tag: false,
+    remove_bad_torrent: false,
+    track_orphans: true,
+  };
+  return {
+    __manageState: manageState,
+    useConfigStore: () => ({
+      getSettingGroup: () => computed(() => manageState),
+    }),
+  };
+});
+
+const AbSettingStub = defineComponent({
+  name: 'AbSettingStub',
+  props: {
+    data: { type: [String, Boolean], default: undefined },
+    description: { type: String, default: '' },
+    label: { type: [String, Function], required: true },
+    prop: { type: Object, default: undefined },
+    type: { type: String, required: true },
+  },
+  emits: ['update:data'],
+  template: '<div class="setting-stub"></div>',
+});
+
+describe('config-manage', () => {
+  it('offers a safe hold default and an explicit higher-revision replacement', async () => {
+    const wrapper = mount(ConfigManage, {
+      global: {
+        stubs: {
+          'ab-fold-panel': { template: '<section><slot /></section>' },
+          'ab-setting': AbSettingStub,
+        },
+      },
+    });
+    const settings = wrapper.findAllComponents(AbSettingStub);
+    const policy = settings.find((setting) => {
+      const label = setting.props('label') as () => string;
+      return label() === 'config.manage_set.revision_conflict_policy';
+    });
+
+    expect(policy).toBeDefined();
+    if (!policy) throw new Error('revision conflict policy setting not found');
+    expect(policy.props('data')).toBe('hold');
+    expect(policy.props('description')).toBe(
+      'config.manage_set.revision_conflict_hint'
+    );
+    expect(policy.props('prop')?.items).toEqual([
+      {
+        id: 1,
+        label: 'config.manage_set.revision_conflict_hold',
+        value: 'hold',
+      },
+      {
+        id: 2,
+        label: 'config.manage_set.revision_conflict_replace',
+        value: 'replace',
+      },
+    ]);
+
+    await policy.vm.$emit('update:data', 'replace');
+    await nextTick();
+    const store = (await import('@/store/config')) as unknown as {
+      __manageState: { revision_conflict_policy: string };
+    };
+    expect(store.__manageState.revision_conflict_policy).toBe('replace');
+  });
+});

webui/src/components/setting/config-manage.vue

@@ -1,14 +1,32 @@
 <script lang="ts" setup>
-import type { BangumiManage, RenameMethod } from '#/config';
-import type { SettingItem } from '#/components';
+import type {
+  BangumiManage,
+  RenameMethod,
+  RevisionConflictPolicy,
+} from '#/config';
+import type { SelectItem, SettingItem } from '#/components';
 
 const { t } = useMyI18n();
 const { getSettingGroup } = useConfigStore();
 
 const manage = getSettingGroup('bangumi_manage');
 const renameMethod: RenameMethod = ['normal', 'pn', 'advance', 'none'];
+const revisionConflictPolicies: RevisionConflictPolicy = ['hold', 'replace'];
 
-const items: SettingItem<BangumiManage>[] = [
+const revisionConflictOptions = computed<SelectItem[]>(() => [
+  {
+    id: 1,
+    label: t('config.manage_set.revision_conflict_hold'),
+    value: revisionConflictPolicies[0],
+  },
+  {
+    id: 2,
+    label: t('config.manage_set.revision_conflict_replace'),
+    value: revisionConflictPolicies[1],
+  },
+]);
+
+const items = computed<SettingItem<BangumiManage>[]>(() => [
   {
     configKey: 'enable',
     label: () => t('config.manage_set.enable'),
@@ -21,6 +39,15 @@ const items: SettingItem<BangumiManage>[] = [
     prop: {
       items: renameMethod,
     },
+  },
+  {
+    configKey: 'revision_conflict_policy',
+    label: () => t('config.manage_set.revision_conflict_policy'),
+    description: t('config.manage_set.revision_conflict_hint'),
+    type: 'select',
+    prop: {
+      items: revisionConflictOptions.value,
+    },
     bottomLine: true,
   },
   {
@@ -43,7 +70,7 @@ const items: SettingItem<BangumiManage>[] = [
     label: () => t('config.manage_set.track_orphans'),
     type: 'switch',
   },
-];
+]);
 </script>
 
 <template>

webui/src/i18n/en.json

@@ -65,6 +65,10 @@
       "eps": "EPS complete",
       "group_tag": "Add Group Tag",
       "method": "Rename Method",
+      "revision_conflict_hint": "When replacement is enabled, the old torrent and its data are deleted only after the higher revision is promoted successfully.",
+      "revision_conflict_hold": "Keep Existing File",
+      "revision_conflict_policy": "Revision Conflict Policy",
+      "revision_conflict_replace": "Replace with Higher Revision",
       "title": "Manage Setting",
       "track_orphans": "Track Unmatched Torrents"
     },
@@ -723,6 +727,10 @@
       "llm_plugin_install_failed": {
         "title": "LLM plugin install failed",
         "body": "{plugin_id}: {message}"
+      },
+      "rename_conflict": {
+        "title": "Media rename conflict",
+        "body": "{torrent_name} → {target_path}: {reason}"
       }
     }
   }

webui/src/i18n/zh-CN.json

@@ -65,6 +65,10 @@
       "eps": "番剧补全",
       "group_tag": "添加组标签",
       "method": "重命名方式",
+      "revision_conflict_hint": "启用替换后，仅在更高修订版成功就位后删除旧种子及其数据。",
+      "revision_conflict_hold": "保留现有文件",
+      "revision_conflict_policy": "修订版冲突处理",
+      "revision_conflict_replace": "替换为更高修订版",
       "title": "番剧管理设置",
       "track_orphans": "记录未匹配种子"
     },
@@ -723,6 +727,10 @@
       "llm_plugin_install_failed": {
         "title": "LLM 插件安装失败",
         "body": "{plugin_id}：{message}"
+      },
+      "rename_conflict": {
+        "title": "媒体文件重命名冲突",
+        "body": "{torrent_name} → {target_path}：{reason}"
       }
     }
   }

webui/src/pages/index/config.vue

@@ -100,7 +100,24 @@ const sections: ConfigSection[] = [
     titleKey: 'config.manage_set.title',
     component: ConfigManage,
     groups: ['bangumi_manage'],
-    keywords: ['rename', 'method', 'eps', 'group', 'tag', 'torrent'],
+    keywords: [
+      'rename',
+      'method',
+      'eps',
+      'group',
+      'tag',
+      'torrent',
+      'revision',
+      'conflict',
+      'version',
+      '修订',
+      '冲突',
+    ],
+    keywordKeys: [
+      'config.manage_set.revision_conflict_policy',
+      'config.manage_set.revision_conflict_hold',
+      'config.manage_set.revision_conflict_replace',
+    ],
   },
   {
     id: 'notification',

webui/src/store/notification.ts

@@ -12,6 +12,7 @@ const KNOWN_KINDS = [
   'update_failed',
   'llm_auth_failure',
   'llm_plugin_install_failed',
+  'rename_conflict',
 ] as const;
 
 export const useNotificationStore = defineStore('notification', () => {

webui/src/test/mocks/api.ts

@@ -135,6 +135,7 @@ export const mockConfig = {
     enable: true,
     eps_complete: false,
     rename_method: 'pn',
+    revision_conflict_policy: 'hold',
     group_tag: false,
     remove_bad_torrent: false,
     track_orphans: true,

webui/types/config.ts

@@ -8,6 +8,8 @@ export type RssParserLang = ['zh', 'en', 'jp'];
 export type RssParserEngine = ['classic', 'tokenizer'];
 /** 重命名方式 */
 export type RenameMethod = ['normal', 'pn', 'advance', 'none'];
+/** 修订版文件名冲突处理策略 */
+export type RevisionConflictPolicy = ['hold', 'replace'];
 /** 代理类型 */
 export type ProxyType = ['http', 'https', 'socks5'];
 /** 通知类型 */
@@ -60,6 +62,7 @@ export interface BangumiManage {
   enable: boolean;
   eps_complete: boolean;
   rename_method: TupleToUnion<RenameMethod>;
+  revision_conflict_policy: TupleToUnion<RevisionConflictPolicy>;
   group_tag: boolean;
   remove_bad_torrent: boolean;
   track_orphans: boolean;
@@ -200,6 +203,7 @@ export const initConfig: Config = {
     enable: true,
     eps_complete: true,
     rename_method: 'normal',
+    revision_conflict_policy: 'hold',
     group_tag: true,
     remove_bad_torrent: true,
     track_orphans: true,