refactor(transfer): make queue admission durable

tests/conftest.py

@@ -230,7 +230,7 @@ def configure_plugin_system_services():
     from app.db.oper.subscribe import SubscribeOper
     from app.db.oper.subscribehistory import SubscribeHistoryOper
     from app.db.oper.transferhistory import TransferHistoryOper
-    from app.db.oper.transferpending import TransferPendingOper
+    from app.db.adapters.transfer import TransactionalTransferAdmissionRepository
     from app.db.oper.user import UserOper
     from app.db.oper.workflow import WorkflowOper, configure_workflow_legacy_writer
     from app.db.oper.message import MessageOper
@@ -303,7 +303,9 @@ def configure_plugin_system_services():
         workflow=lambda: WorkflowOper(),
         download_history=lambda: DownloadHistoryOper(),
         transfer_history=lambda: TransferHistoryOper(),
-        transfer_pending=lambda: TransferPendingOper(),
+        transfer_pending=lambda: TransactionalTransferAdmissionRepository(
+            SessionFactory
+        ),
         media_server=lambda: MediaServerOper(),
         download_failure=lambda: TransactionalDownloadFailureRepository(
             SessionFactory

tests/fixtures/architecture/dependency-baseline.json

@@ -1441,8 +1441,8 @@
       "runtime_only": true
     }
   },
-  "edge_count": 6817,
-  "edge_sha256": "e3d43fec9f7bc936ef5a2ffe7ba11ea054d1ba7d1c42e7101978480cd99a63fe",
+  "edge_count": 6827,
+  "edge_sha256": "34e2be621e40f0f07ff04065655446c4ad0062883701f9e1e2235c29359900f7",
   "edges": [
     "app -> app.runtime",
     "app -> app.runtime.compat",
@@ -3973,6 +3973,8 @@
     "app.application.chain.context -> app.application.configuration",
     "app.application.chain.context -> app.runtime",
     "app.application.chain.context -> app.runtime.stop",
+    "app.application.chain.data -> app.application",
+    "app.application.chain.data -> app.application.transfer",
     "app.application.chain.durable_events -> app.application",
     "app.application.chain.durable_events -> app.application.history",
     "app.application.chain.durable_events -> app.domain",
@@ -5141,6 +5143,14 @@
     "app.db.adapters.subscription -> app.db.uow",
     "app.db.adapters.transaction -> app.db",
     "app.db.adapters.transaction -> app.db.uow",
+    "app.db.adapters.transfer -> app.application",
+    "app.db.adapters.transfer -> app.application.transfer",
+    "app.db.adapters.transfer -> app.db",
+    "app.db.adapters.transfer -> app.db.models",
+    "app.db.adapters.transfer -> app.db.models.transferpending",
+    "app.db.adapters.transfer -> app.db.oper",
+    "app.db.adapters.transfer -> app.db.oper.transferpending",
+    "app.db.adapters.transfer -> app.db.uow",
     "app.db.adapters.workflow -> app.application",
     "app.db.adapters.workflow -> app.application.workflow",
     "app.db.adapters.workflow -> app.db",
@@ -7905,6 +7915,7 @@
     "app.startup.initializers.modules -> app.db.adapters.site",
     "app.startup.initializers.modules -> app.db.adapters.subscription",
     "app.startup.initializers.modules -> app.db.adapters.transaction",
+    "app.startup.initializers.modules -> app.db.adapters.transfer",
     "app.startup.initializers.modules -> app.db.adapters.workflow",
     "app.startup.initializers.modules -> app.db.oper",
     "app.startup.initializers.modules -> app.db.oper.agentchat",
@@ -7919,7 +7930,6 @@
     "app.startup.initializers.modules -> app.db.oper.subscribehistory",
     "app.startup.initializers.modules -> app.db.oper.systemconfig",
     "app.startup.initializers.modules -> app.db.oper.transferhistory",
-    "app.startup.initializers.modules -> app.db.oper.transferpending",
     "app.startup.initializers.modules -> app.db.oper.user",
     "app.startup.initializers.modules -> app.db.oper.userconfig",
     "app.startup.initializers.modules -> app.db.oper.workflow",
@@ -8262,7 +8272,7 @@
     "app.workflow.actions.transfer_file -> app.workflow",
     "app.workflow.actions.transfer_file -> app.workflow.actions"
   ],
-  "module_count": 835,
+  "module_count": 836,
   "modules": [
     "app",
     "app.adapters",
@@ -8660,6 +8670,7 @@
     "app.db.adapters.site",
     "app.db.adapters.subscription",
     "app.db.adapters.transaction",
+    "app.db.adapters.transfer",
     "app.db.adapters.workflow",
     "app.db.base",
     "app.db.decorators",

tests/fixtures/architecture/mypy-baseline.json

@@ -1595,7 +1595,7 @@
     "no-any-return": 3,
     "no-redef": 2,
     "no-untyped-call": 8,
-    "no-untyped-def": 16,
+    "no-untyped-def": 15,
     "operator": 3,
     "return-value": 2,
     "truthy-function": 5,
@@ -1704,9 +1704,6 @@
     "no-untyped-def": 20,
     "type-arg": 1
   },
-  "app/db/models/transferpending.py": {
-    "no-any-return": 1
-  },
   "app/db/models/user.py": {
     "no-untyped-def": 10
   },

tests/fixtures/architecture/ruff-baseline.json

@@ -401,9 +401,6 @@
   "app/application/torrent_cache.py": {
     "I001": 1
   },
-  "app/application/transfer.py": {
-    "I001": 1
-  },
   "app/application/workflow.py": {
     "I001": 1
   },
@@ -1450,9 +1447,6 @@
   "tests/test_interaction_router.py": {
     "E402": 5
   },
-  "tests/test_legacy_import_compat.py": {
-    "I001": 1
-  },
   "tests/test_lifecycle_shutdown.py": {
     "F841": 1
   },
@@ -1873,18 +1867,12 @@
   "tests/test_transfer_movie_collection.py": {
     "I001": 1
   },
-  "tests/test_transfer_pending_replay.py": {
-    "I001": 1
-  },
   "tests/test_transfer_preview.py": {
     "I001": 1
   },
   "tests/test_transfer_queue_count.py": {
     "I001": 1
   },
-  "tests/test_transfer_queue_service.py": {
-    "I001": 1
-  },
   "tests/test_transfer_rename_build_event.py": {
     "I001": 1
   },
@@ -1894,9 +1882,6 @@
   "tests/test_transfer_tmdb_category.py": {
     "I001": 1
   },
-  "tests/test_transfer_worker_lifecycle.py": {
-    "I001": 1
-  },
   "tests/test_transferhistory_media_source_migration.py": {
     "I001": 1
   },

tests/test_architecture_dependencies.py

@@ -478,6 +478,95 @@ def test_transfer_chains_use_explicit_data_port_getters():
     assert violations == []
 
 
+def test_transfer_pending_oper_import_is_confined_to_database_boundary():
+    """宿主仅允许事务适配器和兼容导出直接导入整理待处理 Oper。"""
+    allowed_paths = {
+        "app/db/adapters/transfer.py",
+        "app/db/oper/__init__.py",
+    }
+    violations: list[str] = []
+    for path in APP_ROOT.rglob("*.py"):
+        relative = path.relative_to(PROJECT_ROOT).as_posix()
+        if relative.startswith("app/plugins/") or relative in allowed_paths:
+            continue
+        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Import):
+                if any(
+                    alias.name == "app.db.oper.transferpending"
+                    for alias in node.names
+                ):
+                    violations.append(f"{relative}:{node.lineno}")
+            elif isinstance(node, ast.ImportFrom) and (
+                node.module == "app.db.oper.transferpending"
+                or (
+                    node.module == "app.db.oper"
+                    and any(
+                        alias.name in {"transferpending", "TransferPendingOper"}
+                        for alias in node.names
+                    )
+                )
+            ):
+                violations.append(f"{relative}:{node.lineno}")
+
+    assert violations == []
+
+
+def test_startup_injects_transactional_transfer_admission_repository():
+    """启动组合根必须向 Chain 注入事务型整理准入仓储。"""
+    path = APP_ROOT / "startup" / "initializers" / "modules.py"
+    tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
+    imports_repository = any(
+        isinstance(node, ast.ImportFrom)
+        and node.module == "app.db.adapters.transfer"
+        and any(
+            alias.name == "TransactionalTransferAdmissionRepository"
+            for alias in node.names
+        )
+        for node in ast.walk(tree)
+    )
+    transfer_pending_factories = [
+        keyword.value
+        for node in ast.walk(tree)
+        if isinstance(node, ast.Call)
+        and isinstance(node.func, ast.Name)
+        and node.func.id == "configure_chain_data_ports"
+        for keyword in node.keywords
+        if keyword.arg == "transfer_pending"
+    ]
+
+    assert imports_repository is True
+    assert len(transfer_pending_factories) == 1
+    assert any(
+        isinstance(node, ast.Call)
+        and isinstance(node.func, ast.Name)
+        and node.func.id == "TransactionalTransferAdmissionRepository"
+        for node in ast.walk(transfer_pending_factories[0])
+    )
+
+
+def test_transfer_pending_chain_port_is_typed_without_legacy_proxy():
+    """整理准入端口必须返回明确 Protocol，且旧 Proxy 不得重新出现。"""
+    path = APP_ROOT / "application" / "chain" / "data.py"
+    tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
+    class_names = {
+        node.name
+        for node in tree.body
+        if isinstance(node, ast.ClassDef)
+    }
+    getters = [
+        node
+        for node in tree.body
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
+        and node.name == "get_chain_transfer_pending_port"
+    ]
+
+    assert "TransferPendingPortProxy" not in class_names
+    assert len(getters) == 1
+    assert getters[0].returns is not None
+    assert ast.unparse(getters[0].returns) == "TransferAdmissionRepository"
+
+
 def test_agent_consumers_use_explicit_data_port_getters():
     """Agent 生产模块不得把兼容数据端口代理重新伪装成数据库 Oper。"""
     forbidden = {

tests/test_db_transferpending_queries.py

@@ -6,8 +6,11 @@
 因此这里对着真实数据库断言查回的内容，而不是断言调用了什么。
 """
 import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
 
 from app.db import base as db_base
+from app.db.adapters.transfer import TransactionalTransferAdmissionRepository
 from app.db.models.transferpending import TransferPending
 from app.db.oper.transferpending import TransferPendingOper
 
@@ -195,3 +198,174 @@ def test_oper_discard_and_clear_report_counts(db):
     assert oper.discard(storage="local", src_path="/mnt/a.mkv") == 1
     assert oper.clear() >= 1
     assert oper.list_all() == []
+
+
+def test_stage_admit_is_idempotent_and_keeps_stable_task_id(db):
+    """显式准入重复执行时必须复用首个稳定任务标识。"""
+    first = TransferPending.stage_admit(
+        db.session,
+        task_id="task-first",
+        storage="local",
+        src_path="/mnt/durable.mkv",
+        state="accepted",
+        now_time="2026-08-27 10:00:00",
+    )
+    second = TransferPending.stage_admit(
+        db.session,
+        task_id="task-second",
+        storage="local",
+        src_path="/mnt/durable.mkv",
+        state="accepted",
+        now_time="2026-08-27 11:00:00",
+    )
+
+    assert first is second
+    assert second.task_id == "task-first"
+    assert second.updated_at == "2026-08-27 10:00:00"
+
+
+def test_state_queries_failure_record_and_task_discard(db):
+    """状态查询、失败留痕和按任务删除应共享同一稳定身份。"""
+    TransferPending.stage_admit(
+        db.session,
+        task_id="task-accepted",
+        storage="local",
+        src_path="/mnt/accepted.mkv",
+        state="accepted",
+        now_time="2026-08-27 10:00:00",
+    )
+    db.add(TransferPending(
+        task_id="task-other",
+        storage="local",
+        src_path="/mnt/other.mkv",
+        state="other",
+        created_at="2026-08-27 10:00:01",
+        updated_at="2026-08-27 10:00:01",
+    ))
+    db.session.flush()
+
+    accepted = TransferPending.list_by_state(
+        db.session,
+        state="accepted",
+    )
+    assert [item.task_id for item in accepted] == ["task-accepted"]
+    assert TransferPending.record_enqueue_failure(
+        db.session,
+        task_id="task-accepted",
+        error="queue full",
+        now_time="2026-08-27 10:01:00",
+    ) == 1
+    db.session.expire_all()
+    failed = TransferPending.get_by_identity(
+        db.session,
+        storage="local",
+        src_path="/mnt/accepted.mkv",
+    )
+    assert failed.last_error == "queue full"
+    assert failed.updated_at == "2026-08-27 10:01:00"
+    assert TransferPending.discard_task(
+        db.session,
+        task_id="task-accepted",
+    ) == 1
+
+
+def test_oper_staging_reuses_explicit_write_session(db, monkeypatch):
+    """Oper 的新暂存入口必须服从调用方 Session，不得隐式提交。"""
+    monkeypatch.setattr(
+        db_base,
+        "run_sync_transaction",
+        lambda _operation: (_ for _ in ()).throw(
+            AssertionError("不应创建额外同步事务")
+        ),
+    )
+    oper = TransferPendingOper(db.session)
+
+    pending = oper.stage_admit(
+        task_id="task-explicit",
+        storage="local",
+        src_path="/mnt/explicit-stage.mkv",
+        state="accepted",
+        now_time="2026-08-27 10:00:00",
+    )
+    assert pending.task_id == "task-explicit"
+    assert [item.task_id for item in oper.list_by_state(state="accepted")] == [
+        "task-explicit"
+    ]
+    assert oper.stage_record_enqueue_failure(
+        task_id="task-explicit",
+        error="queue full",
+        now_time="2026-08-27 10:01:00",
+    ) == 1
+    assert oper.stage_discard_task(task_id="task-explicit") == 1
+
+
+def test_transactional_repository_commits_frozen_projections(tmp_path):
+    """适配器应独立提交 UoW，并在会话关闭前冻结应用 DTO。"""
+    engine = create_engine(f"sqlite:///{tmp_path / 'transfer.db'}")
+    TransferPending.__table__.create(engine)
+    factory = sessionmaker(bind=engine)
+    repository = TransactionalTransferAdmissionRepository(factory)
+
+    admitted = repository.admit(
+        storage="local",
+        src_path="/mnt/repository.mkv",
+    )
+    repeated = repository.admit(
+        storage="local",
+        src_path="/mnt/repository.mkv",
+    )
+    assert repeated == admitted
+    assert admitted.task_id
+    assert admitted.state == "accepted"
+    assert repository.list_accepted() == [admitted]
+
+    repository.record_enqueue_failure(
+        task_id=admitted.task_id,
+        error="queue full",
+    )
+    failed = repository.list_accepted()[0]
+    assert failed.last_error == "queue full"
+    assert repository.discard_task(task_id=admitted.task_id) == 1
+    assert repository.list_accepted() == []
+    engine.dispose()
+
+
+def test_transactional_repository_rolls_back_failed_write(monkeypatch):
+    """适配器写入异常时必须回滚自身 UoW 并传播原异常。"""
+    class SessionContext:
+        """为回滚断言提供最小 Session 上下文。"""
+
+        def __init__(self):
+            """初始化提交与回滚计数。"""
+            self.commits = 0
+            self.rollbacks = 0
+
+        def __enter__(self):
+            """返回当前伪会话。"""
+            return self
+
+        def __exit__(self, _exc_type, _exc_value, _traceback):
+            """不吞掉被测异常。"""
+            return False
+
+        def commit(self):
+            """记录提交调用。"""
+            self.commits += 1
+
+        def rollback(self):
+            """记录回滚调用。"""
+            self.rollbacks += 1
+
+    session = SessionContext()
+    repository = TransactionalTransferAdmissionRepository(lambda: session)
+    monkeypatch.setattr(
+        TransferPendingOper,
+        "stage_record_enqueue_failure",
+        lambda self, **_kwargs: (_ for _ in ()).throw(ValueError("write failed")),
+    )
+
+    with pytest.raises(ValueError, match="write failed"):
+        repository.record_enqueue_failure(task_id="task", error="failure")
+
+    assert session.rollbacks == 1
+    assert session.commits == 0

tests/test_legacy_import_compat.py

@@ -14,16 +14,15 @@ from app.runtime.compat.diagnostics import (
 )
 from app.runtime.compat.imports import install_legacy_import_hook
 from app.runtime.compat.manifest import (
+    _MESSAGE_NOTIFICATION_SYMBOL_ALIASES,
     MODULE_ALIASES,
     PACKAGE_ALIASES,
     PACKAGE_EXPORTS,
     SYMBOL_ALIASES,
     VIRTUAL_PACKAGES,
     ModuleAlias,
-    _MESSAGE_NOTIFICATION_SYMBOL_ALIASES,
 )
 
-
 LEGACY_PACKAGE = "legacy_compat_test"
 LEGACY_MODULE = f"{LEGACY_PACKAGE}.target"
 CANONICAL_PACKAGE = "canonical_compat_test"
@@ -295,6 +294,23 @@ def test_physical_modules_resolve_moved_symbols_without_reverse_imports():
     assert schemas_package.TransferQueue is legacy_transfer.TransferQueue
 
 
+def test_legacy_transfer_task_hides_internal_admission_identity():
+    """持久准入身份不得改变插件旧任务字典的公开字段集合。"""
+    legacy_transfer = importlib.import_module("app.sdk._legacy.transfer")
+    task = legacy_transfer.TransferTask(fileitem={
+        "storage": "local",
+        "path": "/downloads/movie.mkv",
+        "type": "file",
+    })
+    public_fields = set(task.to_dict())
+
+    task.bind_admission_task_id("internal-task-id")
+
+    assert set(task.to_dict()) == public_fields
+    assert "task_id" not in task.to_dict()
+    assert "admission_task_id" not in task.to_dict()
+
+
 def test_chain_media_legacy_scraping_symbols_resolve_to_scraping_chain():
     """刮削拆分后，旧 app.chain.media 路径应能继续取用刮削公开符号。"""
     legacy_media = importlib.import_module("app.chain.media")

tests/test_transfer_admission_migration.py

@@ -0,0 +1,241 @@
+"""整理任务持久准入字段的 Alembic 迁移测试。"""
+
+import importlib
+import os
+import uuid
+
+import pytest
+import sqlalchemy as sa
+from alembic.migration import MigrationContext
+from alembic.operations import Operations
+
+from app.db.models.transferpending import TransferPending
+
+try:
+    import psycopg2 as postgres_driver
+    from psycopg2 import sql
+
+    POSTGRESQL_DIALECT = "postgresql+psycopg2"
+except ModuleNotFoundError:
+    import psycopg as postgres_driver
+    from psycopg import sql
+
+    POSTGRESQL_DIALECT = "postgresql+psycopg"
+
+MIGRATION = "database.versions.b1e7d3f5a9c2_3_0_13"
+
+
+def _bind_migration(monkeypatch, connection):
+    """把迁移绑定到隔离数据库连接。"""
+    migration = importlib.import_module(MIGRATION)
+    monkeypatch.setattr(
+        migration,
+        "op",
+        Operations(MigrationContext.configure(connection)),
+    )
+    return migration
+
+
+def _create_legacy_table(connection) -> None:
+    """创建 3.0.12 时代的待整理登记表。"""
+    metadata = sa.MetaData()
+    table = sa.Table(
+        "transferpending",
+        metadata,
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("storage", sa.String(), nullable=False),
+        sa.Column("src_path", sa.String(), nullable=False),
+        sa.Column("created_at", sa.String(), nullable=True),
+    )
+    sa.Index(
+        "ux_transferpending_storage_path",
+        table.c.storage,
+        table.c.src_path,
+        unique=True,
+    )
+    metadata.create_all(connection)
+    connection.execute(table.insert(), [
+        {
+            "id": 1,
+            "storage": "local",
+            "src_path": "/mnt/dated.mkv",
+            "created_at": "2026-08-26 10:00:00",
+        },
+        {
+            "id": 2,
+            "storage": "alist",
+            "src_path": "/mnt/undated.mkv",
+            "created_at": None,
+        },
+    ])
+
+
+def _rows(connection) -> list[dict[str, object]]:
+    """读取迁移后的准入字段快照。"""
+    pending = sa.table(
+        "transferpending",
+        sa.column("id", sa.Integer()),
+        sa.column("task_id", sa.String()),
+        sa.column("state", sa.String()),
+        sa.column("created_at", sa.String()),
+        sa.column("updated_at", sa.String()),
+        sa.column("last_error", sa.Text()),
+    )
+    return [
+        dict(row)
+        for row in connection.execute(
+            sa.select(pending).order_by(pending.c.id)
+        ).mappings().all()
+    ]
+
+
+def test_transfer_admission_upgrade_downgrade_reupgrade(
+    monkeypatch,
+) -> None:
+    """旧行应保守回填，且 SQLite 支持重复升级、降级和再次升级。"""
+    engine = sa.create_engine("sqlite://")
+    with engine.begin() as connection:
+        _create_legacy_table(connection)
+        migration = _bind_migration(monkeypatch, connection)
+
+        migration.upgrade()
+        migration.upgrade()
+
+        inspector = sa.inspect(connection)
+        assert {
+            column["name"]
+            for column in inspector.get_columns("transferpending")
+        } == {column.name for column in TransferPending.__table__.columns}
+        constraints = {
+            constraint["name"]
+            for constraint in inspector.get_unique_constraints("transferpending")
+        }
+        assert "uq_transferpending_task_id" in constraints
+        assert {
+            index["name"]
+            for index in inspector.get_indexes("transferpending")
+        } == {
+            "ix_transferpending_state_created",
+            "ux_transferpending_storage_path",
+        }
+
+        upgraded = _rows(connection)
+        first_task_ids = [row["task_id"] for row in upgraded]
+        assert all(first_task_ids)
+        assert len(set(first_task_ids)) == 2
+        assert {row["state"] for row in upgraded} == {"accepted"}
+        assert upgraded[0]["updated_at"] == upgraded[0]["created_at"]
+        assert upgraded[1]["updated_at"]
+        assert {row["last_error"] for row in upgraded} == {None}
+
+        migration.downgrade()
+        downgraded_inspector = sa.inspect(connection)
+        assert {
+            column["name"]
+            for column in downgraded_inspector.get_columns("transferpending")
+        } == {"id", "storage", "src_path", "created_at"}
+        assert {
+            index["name"]
+            for index in downgraded_inspector.get_indexes("transferpending")
+        } == {"ux_transferpending_storage_path"}
+        legacy_rows = connection.execute(
+            sa.text(
+                "SELECT id, storage, src_path, created_at "
+                "FROM transferpending ORDER BY id"
+            )
+        ).mappings().all()
+        assert [row["src_path"] for row in legacy_rows] == [
+            "/mnt/dated.mkv",
+            "/mnt/undated.mkv",
+        ]
+
+        migration.upgrade()
+        reupgraded = _rows(connection)
+        assert [row["task_id"] for row in reupgraded] == first_task_ids
+        assert {row["state"] for row in reupgraded} == {"accepted"}
+        assert {
+            index["name"]
+            for index in sa.inspect(connection).get_indexes("transferpending")
+        } == {
+            "ix_transferpending_state_created",
+            "ux_transferpending_storage_path",
+        }
+
+
+def test_transfer_admission_migration_runs_on_postgresql(monkeypatch) -> None:
+    """隔离 PostgreSQL 应真实执行准入字段、约束、索引和可逆回滚。"""
+    prefix = "MOVIEPILOT_TEST_POSTGRESQL_"
+    host = os.getenv(f"{prefix}HOST")
+    database = os.getenv(f"{prefix}DATABASE")
+    username = os.getenv(f"{prefix}USERNAME")
+    if not host or not database or not username:
+        pytest.skip("未配置隔离 PostgreSQL migration 测试库")
+
+    port = os.getenv(f"{prefix}PORT", "5432")
+    password = os.getenv(f"{prefix}PASSWORD", "")
+    schema = f"transfer_admission_{uuid.uuid4().hex}"
+    with postgres_driver.connect(
+        host=host,
+        port=port,
+        dbname=database,
+        user=username,
+        password=password,
+    ) as connection:
+        connection.autocommit = True
+        with connection.cursor() as cursor:
+            cursor.execute(sql.SQL("CREATE SCHEMA {}").format(sql.Identifier(schema)))
+
+    engine = None
+    try:
+        engine = sa.create_engine(
+            sa.URL.create(
+                POSTGRESQL_DIALECT,
+                username=username,
+                password=password,
+                host=host,
+                port=int(port),
+                database=database,
+            ),
+            connect_args={"options": f"-csearch_path={schema}"},
+        )
+        with engine.begin() as connection:
+            _create_legacy_table(connection)
+            migration = _bind_migration(monkeypatch, connection)
+            migration.upgrade()
+            migration.upgrade()
+
+            inspector = sa.inspect(connection)
+            assert {
+                constraint["name"]
+                for constraint in inspector.get_unique_constraints(
+                    "transferpending"
+                )
+            } >= {"uq_transferpending_task_id"}
+            assert {
+                index["name"]
+                for index in inspector.get_indexes("transferpending")
+            } >= {"ix_transferpending_state_created"}
+            assert all(row["task_id"] for row in _rows(connection))
+
+            migration.downgrade()
+            assert {
+                column["name"]
+                for column in sa.inspect(connection).get_columns("transferpending")
+            } == {"id", "storage", "src_path", "created_at"}
+    finally:
+        if engine is not None:
+            engine.dispose()
+        with postgres_driver.connect(
+            host=host,
+            port=port,
+            dbname=database,
+            user=username,
+            password=password,
+        ) as connection:
+            connection.autocommit = True
+            with connection.cursor() as cursor:
+                cursor.execute(
+                    sql.SQL("DROP SCHEMA IF EXISTS {} CASCADE").format(
+                        sql.Identifier(schema)
+                    )
+                )

tests/test_transfer_pending_replay.py

@@ -7,26 +7,38 @@
 
 这些测试固定三项不变量：入队即落盘登记、终态即注销、重启能回放。
 """
-from pathlib import Path
 import threading
+from pathlib import Path
 from unittest.mock import MagicMock
 
+from app.application.transfer import TransferAdmission, TransferTask
 from app.chain.transfer import TransferChain
-from app.application.transfer import TransferTask
 from app.schemas.file import FileItem
 
 
-def _build_chain(pendingoper) -> TransferChain:
+def _build_chain(admissions) -> TransferChain:
     """
     构造绕过单例初始化的 TransferChain 骨架。
-    :param pendingoper: 待整理登记管理替身
+    :param admissions: durable admission 仓储替身
     :return: TransferChain 骨架
     """
     chain = object.__new__(TransferChain)
-    chain._pendingoper = pendingoper
+    chain._transfer_admissions = admissions
     return chain
 
 
+def _admission(path: str, task_id: str = "task-1") -> TransferAdmission:
+    """构造一条可脱离数据库会话使用的准入快照。"""
+    return TransferAdmission(
+        task_id=task_id,
+        storage="local",
+        src_path=path,
+        state="accepted",
+        created_at="2026-08-27 10:00:00",
+        updated_at="2026-08-27 10:00:00",
+    )
+
+
 def _task(path: str, storage: str = "local") -> TransferTask:
     """
     构造测试用整理任务。
@@ -45,44 +57,38 @@ def _task(path: str, storage: str = "local") -> TransferTask:
     ))
 
 
-def test_register_pending_records_storage_and_path():
+def test_admit_transfer_records_storage_and_path():
     """
     入队时必须落盘登记「存储 + 源路径」这一最小事实。
     """
-    pendingoper = MagicMock()
-    chain = _build_chain(pendingoper)
+    admissions = MagicMock()
+    admissions.admit.return_value = _admission(
+        "/mnt/cd2/downloads/Movie.2024.mkv"
+    )
+    chain = _build_chain(admissions)
 
-    chain._TransferChain__register_pending(_task("/mnt/cd2/downloads/Movie.2024.mkv"))
-
-    pendingoper.register.assert_called_once_with(
-        storage="local", src_path="/mnt/cd2/downloads/Movie.2024.mkv"
+    result = chain._TransferChain__admit_transfer(
+        _task("/mnt/cd2/downloads/Movie.2024.mkv")
     )
 
-
-def test_register_pending_failure_does_not_break_enqueue():
-    """
-    落盘登记只是重启后的补救手段，登记失败绝不能阻断正常整理。
-    """
-    pendingoper = MagicMock()
-    pendingoper.register.side_effect = RuntimeError("db locked")
-    chain = _build_chain(pendingoper)
-
-    # 不抛异常即为通过
-    chain._TransferChain__register_pending(_task("/mnt/cd2/downloads/Movie.2024.mkv"))
+    admissions.admit.assert_called_once_with(
+        storage="local", src_path="/mnt/cd2/downloads/Movie.2024.mkv"
+    )
+    assert result.task_id == "task-1"
 
 
 def test_discard_pending_on_terminal_state():
     """
     整理到达终态后必须注销登记，否则每次重启都会重复回放。
     """
-    pendingoper = MagicMock()
-    chain = _build_chain(pendingoper)
+    admissions = MagicMock()
+    chain = _build_chain(admissions)
+    task = _task("/mnt/cd2/downloads/Movie.2024.mkv")
+    task.bind_admission_task_id("task-1")
 
-    chain._TransferChain__discard_pending(_task("/mnt/cd2/downloads/Movie.2024.mkv"))
+    chain._TransferChain__discard_pending(task)
 
-    pendingoper.discard.assert_called_once_with(
-        storage="local", src_path="/mnt/cd2/downloads/Movie.2024.mkv"
-    )
+    admissions.discard_task.assert_called_once_with(task_id="task-1")
 
 
 def test_replay_resends_pending_files_to_transfer(tmp_path, monkeypatch):
@@ -92,9 +98,9 @@ def test_replay_resends_pending_files_to_transfer(tmp_path, monkeypatch):
     media = tmp_path / "Movie.2024.mkv"
     media.write_bytes(b"x" * 10)
 
-    pendingoper = MagicMock()
-    pendingoper.list_all.return_value = [("local", str(media))]
-    chain = _build_chain(pendingoper)
+    admissions = MagicMock()
+    admissions.list_accepted.return_value = [_admission(str(media))]
+    chain = _build_chain(admissions)
 
     transferred = []
     monkeypatch.setattr(chain, "do_transfer", lambda **kw: transferred.append(kw["fileitem"]))
@@ -114,16 +120,16 @@ def test_replay_discards_vanished_files(tmp_path):
     """
     源文件已消失的登记要注销，否则每次启动都会重复回放一个不存在的文件。
     """
-    pendingoper = MagicMock()
+    admissions = MagicMock()
     missing = tmp_path / "gone.mkv"
-    pendingoper.list_all.return_value = [("local", str(missing))]
-    chain = _build_chain(pendingoper)
+    admissions.list_accepted.return_value = [_admission(str(missing))]
+    chain = _build_chain(admissions)
     chain.do_transfer = MagicMock()
 
     chain._TransferChain__replay_pending()
 
     chain.do_transfer.assert_not_called()
-    pendingoper.discard.assert_called_once_with(storage="local", src_path=str(missing))
+    admissions.discard_task.assert_called_once_with(task_id="task-1")
 
 
 def test_replay_keeps_registration_when_mount_unreadable(tmp_path, monkeypatch):
@@ -135,9 +141,9 @@ def test_replay_keeps_registration_when_mount_unreadable(tmp_path, monkeypatch):
     media = tmp_path / "Movie.2024.mkv"
     media.write_bytes(b"x")
 
-    pendingoper = MagicMock()
-    pendingoper.list_all.return_value = [("local", str(media))]
-    chain = _build_chain(pendingoper)
+    admissions = MagicMock()
+    admissions.list_accepted.return_value = [_admission(str(media))]
+    chain = _build_chain(admissions)
     chain.do_transfer = MagicMock()
 
     def unreadable(self, *_args, **_kwargs):
@@ -151,7 +157,7 @@ def test_replay_keeps_registration_when_mount_unreadable(tmp_path, monkeypatch):
     chain._TransferChain__replay_pending()
 
     chain.do_transfer.assert_not_called()
-    pendingoper.discard.assert_not_called()
+    admissions.discard_task.assert_not_called()
 
 
 def test_replay_restores_bluray_directory_type(tmp_path, monkeypatch):
@@ -162,9 +168,9 @@ def test_replay_restores_bluray_directory_type(tmp_path, monkeypatch):
     bluray.mkdir()
     src_path = f"{bluray.as_posix()}/"
 
-    pendingoper = MagicMock()
-    pendingoper.list_all.return_value = [("local", src_path)]
-    chain = _build_chain(pendingoper)
+    admissions = MagicMock()
+    admissions.list_accepted.return_value = [_admission(src_path)]
+    chain = _build_chain(admissions)
 
     transferred = []
     monkeypatch.setattr(chain, "do_transfer", lambda **kw: transferred.append(kw["fileitem"]))
@@ -180,9 +186,9 @@ def test_replay_is_noop_without_registrations():
     """
     没有登记时回放不应触碰整理链。
     """
-    pendingoper = MagicMock()
-    pendingoper.list_all.return_value = []
-    chain = _build_chain(pendingoper)
+    admissions = MagicMock()
+    admissions.list_accepted.return_value = []
+    chain = _build_chain(admissions)
     chain.do_transfer = MagicMock()
 
     chain._TransferChain__replay_pending()
@@ -194,9 +200,9 @@ def test_replay_survives_db_failure():
     """
     读取登记失败不能让启动流程报错。
     """
-    pendingoper = MagicMock()
-    pendingoper.list_all.side_effect = RuntimeError("db gone")
-    chain = _build_chain(pendingoper)
+    admissions = MagicMock()
+    admissions.list_accepted.side_effect = RuntimeError("db gone")
+    chain = _build_chain(admissions)
     chain.do_transfer = MagicMock()
 
     chain._TransferChain__replay_pending()
@@ -213,9 +219,12 @@ def test_replay_continues_after_single_file_failure(tmp_path, monkeypatch):
     for item in (first, second):
         item.write_bytes(b"x")
 
-    pendingoper = MagicMock()
-    pendingoper.list_all.return_value = [("local", str(first)), ("local", str(second))]
-    chain = _build_chain(pendingoper)
+    admissions = MagicMock()
+    admissions.list_accepted.return_value = [
+        _admission(str(first), "task-1"),
+        _admission(str(second), "task-2"),
+    ]
+    chain = _build_chain(admissions)
 
     handled = []
 
@@ -241,12 +250,12 @@ def test_replay_stop_keeps_unprocessed_registrations(tmp_path, monkeypatch):
     first = tmp_path / "A.mkv"
     first.write_bytes(b"x")
     missing_second = tmp_path / "gone.mkv"
-    pendingoper = MagicMock()
-    pendingoper.list_all.return_value = [
-        ("local", str(first)),
-        ("local", str(missing_second)),
+    admissions = MagicMock()
+    admissions.list_accepted.return_value = [
+        _admission(str(first), "task-1"),
+        _admission(str(missing_second), "task-2"),
     ]
-    chain = _build_chain(pendingoper)
+    chain = _build_chain(admissions)
     stop_event = threading.Event()
     transferred = []
 
@@ -260,4 +269,4 @@ def test_replay_stop_keeps_unprocessed_registrations(tmp_path, monkeypatch):
     chain._TransferChain__replay_pending(stop_event)
 
     assert transferred == [first.as_posix()]
-    pendingoper.discard.assert_not_called()
+    admissions.discard_task.assert_not_called()

tests/test_transfer_queue_service.py

@@ -1,18 +1,32 @@
-from unittest.mock import Mock
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
 
-from app.application.transfer import TransferQueueService
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from app.application.transfer import TransferAdmission, TransferQueueService
+from app.db.adapters.transfer import TransactionalTransferAdmissionRepository
+from app.db.models.transferpending import TransferPending
 from app.schemas.file import FileItem
-
-from tests.test_transfer_job_manager import make_task
+from tests.test_transfer_job_manager import make_task, make_transfer_chain
 
 
 def _service(**overrides):
     """构造可观测整理队列服务及其默认依赖。"""
     dependencies = {
         "register_task": Mock(return_value=True),
+        "admit_task": Mock(return_value=TransferAdmission(
+            task_id="task-1",
+            storage="local",
+            src_path="/tmp/demo.mkv",
+            state="accepted",
+            created_at="2026-08-27 10:00:00",
+            updated_at="2026-08-27 10:00:00",
+        )),
         "enqueue": Mock(),
         "before_enqueue": Mock(),
-        "after_enqueue": Mock(),
+        "enqueue_failed": Mock(),
         "remove_task": Mock(),
         "list_tasks": Mock(return_value=["job"]),
         "expire_tasks": Mock(),
@@ -22,17 +36,26 @@ def _service(**overrides):
 
 
 def test_transfer_queue_service_put_preserves_registration_order():
-    """入队必须先登记视图，再登记批次、写队列并落盘。"""
+    """入队必须先登记视图和 durable admission，再登记批次并写队列。"""
     calls = []
     service, _ = _service(
         register_task=lambda _task: calls.append("register") or True,
+        admit_task=lambda _task: calls.append("admit") or TransferAdmission(
+            task_id="task-1",
+            storage="local",
+            src_path="/tmp/demo.mkv",
+            state="accepted",
+            created_at="2026-08-27 10:00:00",
+            updated_at="2026-08-27 10:00:00",
+        ),
         before_enqueue=lambda _task: calls.append("batch"),
         enqueue=lambda _item: calls.append("queue"),
-        after_enqueue=lambda _task: calls.append("pending"),
     )
 
-    assert service.put(make_task(1), Mock()) is True
-    assert calls == ["register", "batch", "queue", "pending"]
+    task = make_task(1)
+    assert service.put(task, Mock()) is True
+    assert calls == ["register", "admit", "batch", "queue"]
+    assert task.admission_task_id == "task-1"
 
 
 def test_transfer_queue_service_rejects_duplicate_without_side_effects():
@@ -42,7 +65,81 @@ def test_transfer_queue_service_rejects_duplicate_without_side_effects():
     assert service.put(make_task(1), Mock()) is False
     dependencies["before_enqueue"].assert_not_called()
     dependencies["enqueue"].assert_not_called()
-    dependencies["after_enqueue"].assert_not_called()
+    dependencies["admit_task"].assert_not_called()
+
+
+def test_transfer_queue_service_blocks_enqueue_when_admission_fails():
+    """持久化失败必须撤销作业视图，不能继续加入内存队列。"""
+    service, dependencies = _service(
+        admit_task=Mock(side_effect=RuntimeError("db locked")),
+    )
+    task = make_task(1)
+
+    with pytest.raises(RuntimeError, match="db locked"):
+        service.put(task, Mock())
+
+    dependencies["remove_task"].assert_called_once_with(task.fileitem)
+    dependencies["before_enqueue"].assert_not_called()
+    dependencies["enqueue"].assert_not_called()
+
+
+def test_transfer_queue_service_keeps_admission_when_enqueue_fails():
+    """内存入队失败必须记录原因并清理视图，durable admission 由仓储保留。"""
+    error = RuntimeError("queue closed")
+    service, dependencies = _service(
+        enqueue=Mock(side_effect=error),
+    )
+    task = make_task(1)
+
+    with pytest.raises(RuntimeError, match="queue closed"):
+        service.put(task, Mock())
+
+    dependencies["enqueue_failed"].assert_called_once_with(task, error)
+    dependencies["remove_task"].assert_called_once_with(task.fileitem)
+
+
+def test_transfer_queue_service_cleans_up_when_batch_registration_fails():
+    """准入后的批次登记异常也必须留痕并撤销作业视图。"""
+    error = RuntimeError("batch registration failed")
+    service, dependencies = _service(
+        before_enqueue=Mock(side_effect=error),
+    )
+    task = make_task(1)
+
+    with pytest.raises(RuntimeError, match="batch registration failed"):
+        service.put(task, Mock())
+
+    dependencies["enqueue_failed"].assert_called_once_with(task, error)
+    dependencies["remove_task"].assert_called_once_with(task.fileitem)
+    dependencies["enqueue"].assert_not_called()
+
+
+def test_transfer_queue_service_commits_admission_before_failed_enqueue(tmp_path):
+    """真实仓储已提交后即使内存入队失败，任务也必须带原因留待恢复。"""
+    engine = create_engine(f"sqlite:///{tmp_path / 'durable-admission.db'}")
+    TransferPending.__table__.create(engine)
+    repository = TransactionalTransferAdmissionRepository(sessionmaker(bind=engine))
+    task = make_task(1)
+    service, _ = _service(
+        admit_task=lambda item: repository.admit(
+            storage=item.fileitem.storage,
+            src_path=item.fileitem.path,
+        ),
+        enqueue=Mock(side_effect=RuntimeError("queue closed")),
+        enqueue_failed=lambda item, error: repository.record_enqueue_failure(
+            task_id=item.admission_task_id,
+            error=str(error),
+        ),
+    )
+
+    with pytest.raises(RuntimeError, match="queue closed"):
+        service.put(task, Mock())
+
+    admissions = repository.list_accepted()
+    assert len(admissions) == 1
+    assert admissions[0].task_id == task.admission_task_id
+    assert admissions[0].last_error == "queue closed"
+    engine.dispose()
 
 
 def test_transfer_queue_service_lists_and_removes_through_ports():
@@ -56,3 +153,38 @@ def test_transfer_queue_service_lists_and_removes_through_ports():
     dependencies["expire_tasks"].assert_called_once_with()
     dependencies["list_tasks"].assert_called_once_with()
     dependencies["remove_task"].assert_called_once_with(fileitem)
+
+
+def test_do_transfer_reports_durable_admission_failure():
+    """背景整理准入失败必须返回批次失败，不能伪装成重复任务成功。"""
+    chain = make_transfer_chain()
+    fileitem = make_task(1).fileitem
+    chain._TransferChain__get_trans_fileitems = lambda _item, **_kwargs: [
+        (fileitem, False)
+    ]
+    chain.put_to_queue = Mock(side_effect=RuntimeError("db locked"))
+    no_history = SimpleNamespace(
+        get_by_src=lambda _src, storage=None: None,
+        get_success_by_src=lambda _src, storage=None: None,
+    )
+    no_download = SimpleNamespace(
+        get_by_hash=lambda _hash: None,
+        get_file_by_fullpath=lambda _path: None,
+        get_files_by_savepath=lambda _path: [],
+        get_by_path=lambda _path: None,
+    )
+
+    with patch(
+        "app.chain.transfer.get_chain_transfer_history_port",
+        return_value=no_history,
+    ), patch(
+        "app.chain.transfer.get_chain_download_history_port",
+        return_value=no_download,
+    ), patch(
+        "app.chain.transfer.get_configured_system_config",
+        return_value=SimpleNamespace(get=lambda _key: None),
+    ):
+        state, message = chain.do_transfer(fileitem=fileitem, background=True)
+
+    assert state is False
+    assert "加入整理队列失败：db locked" in message

tests/test_transfer_worker_lifecycle.py

@@ -10,10 +10,10 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
+from app.application.transfer import TransferAdmission, TransferQueue, TransferTask
 from app.chain.transfer import TransferChain
 from app.foundation.singleton import Singleton
 from app.runtime.config import global_vars
-from app.application.transfer import TransferQueue, TransferTask
 from app.schemas.file import FileItem
 from app.startup.initializers import transfer as transfer_initializer
 
@@ -411,6 +411,62 @@ def test_worker_settles_progress_when_only_stop_sentinel_remains(monkeypatch) ->
         assert list(chain._queue.queue) == [chain._QUEUE_STOP_SENTINEL]
 
 
+def test_durable_task_identity_flows_from_queue_to_terminal_discard(monkeypatch) -> None:
+    """准入生成的稳定身份必须随队列任务到 worker 终态并准确注销。"""
+    chain = _build_chain()
+    chain.runtime_config.transfer_task_timeout = 0
+    task = TransferTask(fileitem=FileItem(
+        storage="local",
+        path="/downloads/durable.mkv",
+        type="file",
+        name="durable.mkv",
+        basename="durable",
+        extension="mkv",
+    ))
+    discarded = threading.Event()
+    admissions = MagicMock()
+    admissions.admit.return_value = TransferAdmission(
+        task_id="durable-task-id",
+        storage="local",
+        src_path=task.fileitem.path,
+        state="accepted",
+        created_at="2026-08-27 10:00:00",
+        updated_at="2026-08-27 10:00:00",
+    )
+    admissions.discard_task.side_effect = (
+        lambda **_kwargs: discarded.set() or 1
+    )
+    chain._transfer_admissions = admissions
+    chain.jobview = MagicMock()
+    chain.jobview.add_task.return_value = True
+    chain.jobview.pending_total.return_value = 1
+    chain._register_scrape_batch_task = MagicMock()
+    chain._finish_scrape_batch_task = MagicMock()
+    chain._progress = MagicMock()
+    chain._active_tasks = 0
+    chain._processed_num = 0
+    chain._fail_num = 0
+    chain._total_num = 0
+    chain._TransferChain__handle_transfer = MagicMock(return_value=(True, ""))
+    monkeypatch.setattr(global_vars, "STOP_EVENT", threading.Event())
+
+    assert chain.put_to_queue(task) is True
+    stop_event = threading.Event()
+    worker = threading.Thread(
+        target=chain._TransferChain__start_transfer,
+        args=(stop_event,),
+        daemon=True,
+    )
+    worker.start()
+    assert discarded.wait(timeout=1)
+    stop_event.set()
+    worker.join(timeout=1)
+
+    assert worker.is_alive() is False
+    assert task.admission_task_id == "durable-task-id"
+    admissions.discard_task.assert_called_once_with(task_id="durable-task-id")
+
+
 def test_claimed_task_prevents_progress_settlement_before_active_registration() -> None:
     """其他 worker 已取走真实任务但尚未登记 active 时，当前批次不得提前结算。"""
     chain = _build_chain()