fix(douyu): 替换 CK 是最后一步，只有登录成功才写回记录

上一版把可见性改成「记录里还有 CK」，账号不会再消失，但根因还在：批次启动
时就把复用的成功记录改成 pending（旧 CK 仍在），停止或失败后没人恢复状态，
记录一直显示成一次没完成的登录。

按「替换 CK 是最后一步」把两者拆开：

- login_tasks 新增 attempt_status / attempt_message / attempt_batch_id /
  attempt_at，批次只写 attempt_*，status/cookie/message 只在登录成功时更新；
- _begin_attempt 只登记尝试，_mark_attempt_failed 只记录失败（已有 CK 的记录
  原样不动），_write_login_success 是唯一替换旧 CK 的入口；
- 停止批次、代理不可用、登录异常、进程重启清理全部改为收敛 attempt_*，
  因此停止一个批次不可能再让账号失去 CK；
- 复用的记录改用「持有 CK 的记录」，不再创建第二条记录，失败记录在无 CK 时
  才复用；
- 登录任务列表/统计按 attempt_* 展示「这次登录怎么样了」，批次筛选同时匹配
  attempt_batch_id；CK 操作页的重登状态也从 attempt_* 映射；
- 迁移 20261001_0035 归一存量数据：持有 CK 的记录 status 统一为 success、
  历史状态搬进 attempt_*，无 CK 且卡在 pending/running 的收敛为 failed。

新增测试：批次失败/停止不碰旧 CK、成功后替换 CK、列表展示尝试状态、
迁移存量数据归一。

tests/test_cookie_operations.py

@@ -207,8 +207,8 @@ class TestCookieOperation:
             )
 
     def test_relogin_failed_record_stays_visible_and_can_be_checked(self):
-        self.owned_task.status = "relogin_failed"
-        self.owned_task.message = "重新登录失败: 密码错误（旧 Cookie 已保留）"
+        self.owned_task.attempt_status = "failed"
+        self.owned_task.attempt_message = "重新登录失败: 密码错误（旧 Cookie 已保留）"
         self.session.commit()
 
         result = list_cookie_operations(
@@ -223,9 +223,13 @@ class TestCookieOperation:
         assert result["total"] == 1
         assert result["items"][0]["relogin_status"] == "relogin_failed"
         assert "旧 Cookie 已保留" in result["items"][0]["relogin_message"]
+        # 失败的重登尝试不影响记录持有的 CK
+        assert self.owned_task.cookie == "owned-secret"
+        assert self.owned_task.status == "success"
 
     def test_service_restart_cleans_orphan_relogin_tasks(self):
-        self.owned_task.status = "relogin_running"
+        self.owned_task.attempt_status = "running"
+        self.owned_task.attempt_message = "正在重新登录，旧 Cookie 保留中"
         self.session.commit()
 
         cleaned = cleanup_orphan_relogin_tasks(self.session)
@@ -233,13 +237,17 @@ class TestCookieOperation:
         assert cleaned == 1
         self.session.refresh(self.owned_task)
         self.session.refresh(self.other_task)
-        assert self.owned_task.status == "relogin_failed"
-        assert "服务重启" in self.owned_task.message
+        # 只收敛尝试状态，已有 CK 与成功状态原封不动
+        assert self.owned_task.attempt_status == "failed"
+        assert "服务重启" in self.owned_task.attempt_message
+        assert self.owned_task.status == "success"
+        assert self.owned_task.cookie == "owned-secret"
+        assert self.other_task.attempt_status == ""
         assert self.other_task.status == "success"
 
     def test_service_restart_cleans_orphan_login_tasks(self):
-        """普通登录批次遗留的 pending/running 必须收敛，且保留旧 CK。"""
-        self.owned_task.status = "pending"
+        """普通登录批次遗留的在途尝试必须收敛，且保留旧 CK。"""
+        self.owned_task.attempt_status = "pending"
         self.owned_task.cookie = "acf_uid=owned"
         self.session.commit()
 
@@ -248,11 +256,25 @@ class TestCookieOperation:
         assert cleaned == 1
         self.session.refresh(self.owned_task)
         self.session.refresh(self.other_task)
-        assert self.owned_task.status == "failed"
+        assert self.owned_task.attempt_status == "failed"
         assert self.owned_task.cookie == "acf_uid=owned"
-        assert "服务重启" in self.owned_task.message
+        assert self.owned_task.status == "success"
+        assert "服务重启" in self.owned_task.attempt_message
         assert self.other_task.status == "success"
 
+    def test_service_restart_cleans_legacy_in_flight_status(self):
+        """升级前直接把状态写在 status 上的在途记录也要能收敛。"""
+        self.owned_task.status = "relogin_running"
+        self.owned_task.attempt_status = ""
+        self.session.commit()
+
+        cleaned = cleanup_orphan_login_tasks(self.session)
+
+        assert cleaned == 1
+        self.session.refresh(self.owned_task)
+        assert self.owned_task.attempt_status == "failed"
+        assert self.owned_task.cookie == "owned-secret"
+
     def test_runner_copies_proxy_config_before_background_execution(self):
         proxy = ProxyConfig(
             enabled=True,

tests/test_login_attempt_isolation.py

@@ -0,0 +1,224 @@
+"""回归：替换 CK 必须是最后一步，只有登录成功才会写回记录。
+
+事故背景：批次启动时会把复用的成功记录直接改成 pending（旧 CK 仍保留），
+批次被停止或失败后没有任何地方把它改回成功，账号于是从工作台消失。
+现在尝试状态独立存放在 attempt_*，status/cookie/message 只在登录成功时
+更新，所以停止、失败、进程重启都不可能破坏账号已有的 CK。
+"""
+
+import tempfile
+from pathlib import Path
+from types import SimpleNamespace
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+import web.backend.database as database_module
+from web.backend.database import Base
+from web.backend.models import Account, LoginTask, User
+from web.backend.routers.login import list_tasks
+from web.backend.services import login_service
+
+
+class _FakeLogin:
+    """替身登录器：不发网络请求，结果由类属性控制。"""
+
+    success = True
+    cookie = "new-cookie"
+    message = "登录成功"
+    code = None
+    calls = 0
+
+    def __init__(self, *_args, **_kwargs):
+        pass
+
+    def login(self):
+        type(self).calls += 1
+        return SimpleNamespace(
+            success=type(self).success,
+            cookie=type(self).cookie if type(self).success else "",
+            message=type(self).message,
+            code=type(self).code,
+        )
+
+    def check_account(self):
+        return self.login()
+
+
+class TestLoginAttemptIsolation:
+    def setup_method(self):
+        self.tmpdir = tempfile.TemporaryDirectory()
+        url = f"sqlite:///{Path(self.tmpdir.name) / 'attempt.db'}"
+        self.engine = create_engine(url, connect_args={"check_same_thread": False})
+        Base.metadata.create_all(self.engine)
+        # 批次执行器在后台线程里用 SessionLocal，测试期间指向同一个文件库。
+        database_module.SessionLocal.configure(bind=self.engine)
+        self.session = sessionmaker(bind=self.engine)()
+
+        self.user = User(username="admin", password_hash="hash", role="super_admin")
+        self.session.add(self.user)
+        self.session.flush()
+        self.account = Account(
+            username="远行者103",
+            password="password",
+            email="yuanxingzhe@example.com",
+            email_password="mail-password",
+        )
+        self.session.add(self.account)
+        self.session.flush()
+        self.task = LoginTask(
+            batch_id="old-batch",
+            account_id=self.account.id,
+            created_by=self.user.id,
+            status="success",
+            cookie="old-cookie",
+            message="登录成功",
+        )
+        self.session.add(self.task)
+        self.session.commit()
+        self.task_id = self.task.id
+
+        _FakeLogin.success = True
+        _FakeLogin.cookie = "new-cookie"
+        _FakeLogin.message = "登录成功"
+        _FakeLogin.calls = 0
+
+    def teardown_method(self):
+        database_module.SessionLocal.configure(bind=database_module.engine)
+        self.session.close()
+        Base.metadata.drop_all(self.engine)
+        self.engine.dispose()
+        self.tmpdir.cleanup()
+
+    def _run_batch(self, monkeypatch, *, mode="login", account_ids=None, stop=False):
+        monkeypatch.setattr(login_service, "DouyuLogin", _FakeLogin)
+        # 批次结束时 runner 会关掉自己的会话，因此单独给它一个。
+        runner_db = sessionmaker(bind=self.engine)()
+        runner = login_service.LoginBatchRunner(
+            db=runner_db,
+            account_ids=account_ids if account_ids is not None else [self.account.id],
+            created_by=self.user.id,
+            creator_permissions=["login:view_all"],
+            proxy_config=None,
+            concurrency=1,
+            mode=mode,
+        )
+        if stop:
+            runner.stop()
+        runner.run()
+        return runner
+
+    def _fresh(self, task_id: int) -> LoginTask:
+        session = sessionmaker(bind=self.engine)()
+        try:
+            return session.query(LoginTask).filter(LoginTask.id == task_id).one()
+        finally:
+            session.close()
+
+    def _fresh_user(self, session) -> User:
+        return session.query(User).filter(User.id == self.user.id).one()
+
+    def test_failed_attempt_keeps_existing_cookie(self, monkeypatch):
+        _FakeLogin.success = False
+        _FakeLogin.message = "账号密码错误"
+
+        self._run_batch(monkeypatch)
+
+        task = self._fresh(self.task_id)
+        # 记录持有 CK 的字段完全没被碰过
+        assert task.status == "success"
+        assert task.cookie == "old-cookie"
+        assert task.message == "登录成功"
+        assert task.batch_id == "old-batch"
+        # 失败只体现在尝试字段上
+        assert task.attempt_status == "failed"
+        assert task.attempt_message == "账号密码错误"
+        assert task.attempt_at is not None
+
+    def test_failed_attempt_reuses_the_cookie_row(self, monkeypatch):
+        _FakeLogin.success = False
+        _FakeLogin.message = "账号密码错误"
+
+        self._run_batch(monkeypatch)
+
+        session = sessionmaker(bind=self.engine)()
+        try:
+            rows = (
+                session.query(LoginTask)
+                .filter(LoginTask.account_id == self.account.id)
+                .all()
+            )
+        finally:
+            session.close()
+        assert [row.id for row in rows] == [self.task_id]
+
+    def test_successful_attempt_replaces_cookie(self, monkeypatch):
+        self._run_batch(monkeypatch)
+
+        task = self._fresh(self.task_id)
+        assert task.status == "success"
+        assert task.cookie == "new-cookie"
+        assert task.message == "登录成功"
+        assert task.attempt_status == "success"
+        assert task.attempt_at is not None
+        assert task.batch_id != "old-batch"
+
+    def test_stopped_batch_keeps_existing_cookie(self, monkeypatch):
+        self._run_batch(monkeypatch, stop=True)
+
+        task = self._fresh(self.task_id)
+        assert _FakeLogin.calls == 0
+        assert task.cookie == "old-cookie"
+        assert task.status == "success"
+        assert task.attempt_status == "failed"
+        assert "旧 Cookie 已保留" in task.attempt_message
+
+    def test_list_tasks_reports_attempt_state_without_losing_ck_row(self, monkeypatch):
+        _FakeLogin.success = False
+        _FakeLogin.message = "账号密码错误"
+
+        runner = self._run_batch(monkeypatch)
+
+        read_db = sessionmaker(bind=self.engine)()
+        try:
+            rows = list_tasks(
+                batch_id=runner.batch_id, db=read_db, current=self._fresh_user(read_db)
+            )
+        finally:
+            read_db.close()
+        assert len(rows) == 1
+        row = rows[0]
+        # 页面看到的是「这次尝试怎么样了」，而记录仍然持有 CK
+        assert row.status == "failed"
+        assert row.message == "账号密码错误"
+        assert row.batch_id == runner.batch_id
+        assert self._fresh(self.task_id).cookie == "old-cookie"
+
+    def test_account_without_cookie_still_reports_failure(self, monkeypatch):
+        """首次登录失败的账号没有 CK 需要保护，保持原来的失败语义。"""
+        fresh_account = Account(
+            username="no-cookie",
+            password="password",
+            email="nocookie@example.com",
+            email_password="mail-password",
+        )
+        self.session.add(fresh_account)
+        self.session.commit()
+        _FakeLogin.success = False
+        _FakeLogin.message = "账号密码错误"
+
+        self._run_batch(monkeypatch, account_ids=[fresh_account.id])
+
+        session = sessionmaker(bind=self.engine)()
+        try:
+            task = (
+                session.query(LoginTask)
+                .filter(LoginTask.account_id == fresh_account.id)
+                .one()
+            )
+            assert task.status == "failed"
+            assert task.message == "账号密码错误"
+            assert task.attempt_status == "failed"
+            assert task.cookie == ""
+        finally:
+            session.close()

tests/test_login_attempt_migration.py

@@ -0,0 +1,152 @@
+"""迁移 20261001_0035 的存量数据归一验证。
+
+线上升级时要保证：
+- 持有 CK 的记录 status 归一到 success，历史批次状态搬进 attempt_*；
+- 没有 CK 且卡在 pending/running 的记录收敛为 failed；
+- 任何情况下都不丢 CK。
+"""
+
+import tempfile
+from pathlib import Path
+
+from alembic import command
+from alembic.config import Config
+from sqlalchemy import create_engine, text
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+
+
+def _migration_config(url: str) -> Config:
+    config = Config(str(PROJECT_ROOT / "alembic.ini"))
+    config.set_main_option(
+        "script_location", str(PROJECT_ROOT / "web" / "backend" / "migrations")
+    )
+    config.set_main_option("sqlalchemy.url", url)
+    return config
+
+
+def _insert_task(conn, *, status: str, cookie: str, message: str = "") -> None:
+    conn.execute(
+        text(
+            """
+            INSERT INTO login_tasks
+                (batch_id, account_id, status, cookie, message, created_by,
+                 created_at, finished_at)
+            VALUES
+                ('batch-1', 1, :status, :cookie, :message, 1,
+                 '2026-09-30 00:00:00', '2026-09-30 01:00:00')
+            """
+        ),
+        {"status": status, "cookie": cookie, "message": message},
+    )
+
+
+class TestLoginAttemptMigration:
+    def setup_method(self):
+        import web.backend.database as database_module
+
+        self.database_module = database_module
+        self.original_url = database_module.DATABASE_URL
+        self.tmpdir = tempfile.TemporaryDirectory()
+        self.url = f"sqlite:///{Path(self.tmpdir.name) / 'upgrade.db'}"
+        # env.py 用模块级 DATABASE_URL 覆盖配置里的连接串。
+        database_module.DATABASE_URL = self.url
+        self.config = _migration_config(self.url)
+        command.upgrade(self.config, "20260901_0034")
+
+        self.engine = create_engine(self.url)
+        with self.engine.begin() as conn:
+            conn.execute(
+                text(
+                    """
+                    INSERT INTO accounts
+                        (id, username, password, email, email_password,
+                         email_imap_server, email_imap_port, email_imap_ssl,
+                         tag, remark, uid, nickname, created_at, updated_at)
+                    VALUES
+                        (1, '远行者103', 'p', 'e@example.com', 'm',
+                         '', 993, 1, '', '', '', '', '2026-09-30 00:00:00',
+                         '2026-09-30 00:00:00')
+                    """
+                )
+            )
+            conn.execute(
+                text(
+                    """
+                    INSERT INTO users (id, username, password_hash, role, is_active)
+                    VALUES (1, 'admin', 'hash', 'super_admin', 1)
+                    """
+                )
+            )
+            _insert_task(conn, status="success", cookie="ck-success")
+            _insert_task(
+                conn,
+                status="relogin_failed",
+                cookie="ck-relogin",
+                message="重新登录失败: 密码错误（旧 Cookie 已保留）",
+            )
+            _insert_task(conn, status="pending", cookie="ck-pending")
+            _insert_task(conn, status="pending", cookie="")
+            _insert_task(conn, status="running", cookie="")
+            _insert_task(conn, status="failed", cookie="ck-failed", message="登录失败")
+
+    def teardown_method(self):
+        self.database_module.DATABASE_URL = self.original_url
+        self.engine.dispose()
+        self.tmpdir.cleanup()
+
+    def _rows(self) -> list[dict]:
+        with self.engine.connect() as conn:
+            rows = conn.execute(
+                text(
+                    """
+                    SELECT status, cookie, message, attempt_status,
+                           attempt_message, attempt_batch_id, attempt_at
+                      FROM login_tasks
+                     ORDER BY id
+                    """
+                )
+            ).mappings()
+            return [dict(row) for row in rows]
+
+    def test_upgrade_normalizes_existing_rows(self):
+        command.upgrade(self.config, "head")
+
+        rows = self._rows()
+        (
+            success,
+            relogin_failed,
+            pending_with_ck,
+            pending_without_ck,
+            running_without_ck,
+            failed_with_ck,
+        ) = rows
+
+        # 已经成功的记录：status 不变，也没有历史尝试需要搬迁
+        assert success["status"] == "success"
+        assert success["cookie"] == "ck-success"
+        assert success["attempt_status"] in (None, "")
+
+        # 重登失败：CK 保留，status 归一为 success，失败信息进 attempt_*
+        assert relogin_failed["cookie"] == "ck-relogin"
+        assert relogin_failed["status"] == "success"
+        assert relogin_failed["attempt_status"] == "failed"
+        assert "旧 Cookie 已保留" in relogin_failed["attempt_message"]
+        assert relogin_failed["attempt_batch_id"] == "batch-1"
+
+        # 卡在 pending 但持有 CK：同样归一为 success + 中断说明
+        assert pending_with_ck["cookie"] == "ck-pending"
+        assert pending_with_ck["status"] == "success"
+        assert pending_with_ck["attempt_status"] == "failed"
+        assert "旧 Cookie 已保留" in pending_with_ck["attempt_message"]
+
+        # 没有 CK 的在途记录：收敛为失败，不再永久排队
+        for row in (pending_without_ck, running_without_ck):
+            assert row["status"] == "failed"
+            assert row["attempt_status"] == "failed"
+            assert "服务重启" in row["message"]
+
+        # 复用记录上的旧 CK 不能被清掉
+        assert failed_with_ck["status"] == "success"
+        assert failed_with_ck["cookie"] == "ck-failed"
+        assert failed_with_ck["attempt_status"] == "failed"

tests/test_migrations.py

@@ -22,7 +22,7 @@ TESTS_DIR = Path(__file__).resolve().parent
 PROJECT_ROOT = TESTS_DIR.parent
 VERSIONS_DIR = PROJECT_ROOT / "web" / "backend" / "migrations" / "versions"
 
-HEAD_REVISION = "20260901_0034"
+HEAD_REVISION = "20261001_0035"
 
 # 迁移新增、但不声明在模型里的复合查询索引。
 EXTRA_INDEXES = (

web/backend/migrations/versions/20261001_0035_login_attempt_fields.py

@@ -0,0 +1,139 @@
+"""login_tasks 增加 attempt_* 字段：登录尝试与 CK 记录分离。
+
+事故背景：批次启动时会把复用的成功记录直接改成 pending（旧 CK 仍保留），
+批次被停止或失败后没有任何地方把它改回成功，账号于是从工作台消失。
+按「替换 CK 是最后一步、只有成功后才会替换旧的 CK」的原则，把尝试状态独立
+出来：批次只写 attempt_*，只有登录成功才会把新 CK 写回 status/cookie/message，
+因此停止、失败、进程重启都不可能破坏已有 CK。
+
+存量数据归一：
+- 持有 CK 但状态不是 success 的记录（relogin_failed/failed/pending/running…）
+  把状态、消息、批次、时间搬到 attempt_*，status 归一到 success，让 status
+  重新只表示「这条记录持有可用 CK」；
+- 没有 CK 且卡在 pending/running 的记录收敛为 failed，避免永久显示排队中。
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "20261001_0035"
+down_revision: str | None = "20260901_0034"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+# 升级前可能停留在这些状态、但旧 CK 已经保留在记录里的批次状态。
+_INTERRUPTED_STATUSES = ("pending", "running", "relogin_pending", "relogin_running")
+_INTERRUPTED_SQL = "(" + ",".join(f"'{item}'" for item in _INTERRUPTED_STATUSES) + ")"
+# 历史重登状态统一归一为 failed，避免 attempt_status 里混入页面不认识的值。
+_LEGACY_ATTEMPT_STATUSES = _INTERRUPTED_STATUSES + ("relogin_failed",)
+_LEGACY_ATTEMPT_SQL = (
+    "(" + ",".join(f"'{item}'" for item in _LEGACY_ATTEMPT_STATUSES) + ")"
+)
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    columns = {item["name"] for item in sa.inspect(bind).get_columns("login_tasks")}
+    if "attempt_status" not in columns:
+        # NOT NULL + server_default 会让存量行直接填上空串，避免出现 NULL
+        # 导致 NOT IN 查询把历史记录整批排除。
+        op.add_column(
+            "login_tasks",
+            sa.Column(
+                "attempt_status",
+                sa.String(length=32),
+                nullable=False,
+                server_default="",
+            ),
+        )
+    if "attempt_message" not in columns:
+        op.add_column(
+            "login_tasks",
+            sa.Column(
+                "attempt_message",
+                sa.String(length=512),
+                nullable=False,
+                server_default="",
+            ),
+        )
+    if "attempt_batch_id" not in columns:
+        op.add_column(
+            "login_tasks",
+            sa.Column(
+                "attempt_batch_id",
+                sa.String(length=64),
+                nullable=False,
+                server_default="",
+            ),
+        )
+    if "attempt_at" not in columns:
+        op.add_column(
+            "login_tasks",
+            sa.Column("attempt_at", sa.DateTime(), nullable=True),
+        )
+
+    # 1) 持有 CK 的记录：历史批次状态搬进 attempt_*（status 稍后再归一，
+    #    避免同一条 UPDATE 里读到被覆盖后的值）。
+    bind.execute(
+        sa.text(
+            f"""
+            UPDATE login_tasks
+               SET attempt_status = CASE
+                       WHEN status IN {_LEGACY_ATTEMPT_SQL} THEN 'failed'
+                       ELSE status
+                   END,
+                   attempt_message = CASE
+                       WHEN status IN {_INTERRUPTED_SQL}
+                           THEN '登录已中断（旧 Cookie 已保留）'
+                       ELSE COALESCE(message, '')
+                   END,
+                   attempt_batch_id = COALESCE(batch_id, ''),
+                   attempt_at = COALESCE(finished_at, created_at)
+             WHERE cookie IS NOT NULL
+               AND cookie <> ''
+               AND status <> 'success'
+               AND (attempt_status IS NULL OR attempt_status = '')
+            """
+        )
+    )
+
+    # 2) 持有 CK 的记录：status 只表示「持有可用 CK」。
+    bind.execute(
+        sa.text(
+            """
+            UPDATE login_tasks
+               SET status = 'success'
+             WHERE cookie IS NOT NULL
+               AND cookie <> ''
+               AND status <> 'success'
+            """
+        )
+    )
+
+    # 3) 没有 CK 且卡在批次中的记录：收敛为失败，不再永久排队中。
+    bind.execute(
+        sa.text(
+            f"""
+            UPDATE login_tasks
+               SET status = 'failed',
+                   message = '登录已中断（服务重启），无 Cookie',
+                   finished_at = COALESCE(finished_at, created_at),
+                   attempt_status = 'failed',
+                   attempt_message = '登录已中断（服务重启），无 Cookie',
+                   attempt_at = COALESCE(attempt_at, created_at)
+             WHERE status IN {_INTERRUPTED_SQL}
+               AND (cookie IS NULL OR cookie = '')
+               AND (attempt_status IS NULL OR attempt_status = '')
+            """
+        )
+    )
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    columns = {item["name"] for item in sa.inspect(bind).get_columns("login_tasks")}
+    for name in ("attempt_at", "attempt_batch_id", "attempt_message", "attempt_status"):
+        if name in columns:
+            op.drop_column("login_tasks", name)

web/backend/models.py

@@ -155,6 +155,18 @@ class LoginTask(Base):
         JSON, nullable=True
     )  # {fish_ball, nickname, level, message}
     ck_checked_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    # 最近一次登录尝试的状态。批次只写 attempt_*，只有登录成功才会把新 CK
+    # 写回 status/cookie/message，因此停止、失败、进程重启都不会破坏已有 CK。
+    attempt_status: Mapped[str] = mapped_column(
+        String(32), default="", server_default=""
+    )  # pending / running / success / failed / stopped / 空=没有进行过的尝试
+    attempt_message: Mapped[str] = mapped_column(
+        String(512), default="", server_default=""
+    )
+    attempt_batch_id: Mapped[str] = mapped_column(
+        String(64), default="", server_default=""
+    )
+    attempt_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
 
     account: Mapped[Account] = relationship("Account", back_populates="login_tasks")
 

web/backend/routers/cookies.py

@@ -19,7 +19,7 @@ from ..permissions import get_user_permissions, user_has_permission
 from ..schemas import CookieReloginRequest
 from ..services.cookie_check_service import check_douyu_cookie
 from ..services.douyu_service import (
-    LOGIN_IN_FLIGHT_STATUSES,
+    LOGIN_ATTEMPT_IN_FLIGHT_STATUSES,
     cookie_holding_filter,
     cookie_operable_filter,
 )
@@ -76,6 +76,34 @@ def _order_cookie_tasks(query, selected_names: list[str]):
     return query.order_by(LoginTask.finished_at.desc(), LoginTask.id.desc())
 
 
+# CK 操作页沿用的重登状态词表；attempt_* 是唯一来源，历史值一并兼容。
+_RELOGIN_STATUS_BY_ATTEMPT = {
+    "pending": "relogin_pending",
+    "running": "relogin_running",
+    "failed": "relogin_failed",
+    "error": "relogin_failed",
+    "relogin_pending": "relogin_pending",
+    "relogin_running": "relogin_running",
+    "relogin_failed": "relogin_failed",
+}
+
+
+def _relogin_display(task: LoginTask) -> dict:
+    """把最近一次登录尝试映射成 CK 操作页展示用的重登状态。
+
+    账号持有的 CK 与「这次尝试的结果」分开存储，页面只关心后者；尝试结束
+    后旧 CK 仍在记录里，因此停止或失败都不会让这一行消失。
+    """
+    attempt_status = task.attempt_status or ""
+    return {
+        "relogin_status": _RELOGIN_STATUS_BY_ATTEMPT.get(attempt_status, ""),
+        "relogin_message": task.attempt_message or "",
+        "relogin_batch_id": task.attempt_batch_id
+        if attempt_status in LOGIN_ATTEMPT_IN_FLIGHT_STATUSES
+        else "",
+    }
+
+
 def _check_one_cookie(task: LoginTask) -> dict:
     """将共享检测结果补充为 Cookie 记录的接口返回格式。"""
     result = check_douyu_cookie(task.cookie or "")
@@ -291,11 +319,7 @@ def list_cookie_operations(
                 "ck_check_status": task.ck_check_status or "",
                 "ck_checked_at": _fmt_dt(task.ck_checked_at),
                 "created_at": _fmt_dt(task.finished_at),
-                "relogin_status": task.status if task.status != "success" else "",
-                "relogin_message": task.message or "",
-                "relogin_batch_id": task.batch_id
-                if task.status in {"relogin_pending", "relogin_running"}
-                else "",
+                **_relogin_display(task),
             }
             for task in tasks
         ],
@@ -575,17 +599,16 @@ def _start_relogin_tasks(
             skipped.append(task.id)
             continue
         if not acc.password or not acc.email:
-            task.status = "relogin_failed"
-            task.message = "重新登录失败: 账号缺少密码或邮箱，旧 Cookie 已保留"
-            task.finished_at = datetime.now(UTC)
+            # 只记录这次尝试失败，账号已有的 CK 原封不动。
+            task.attempt_status = "failed"
+            task.attempt_message = "重新登录失败: 账号缺少密码或邮箱，旧 Cookie 已保留"
+            task.attempt_at = datetime.now(UTC)
             skipped.append(task.id)
             continue
-        if task.status in LOGIN_IN_FLIGHT_STATUSES:
+        if task.attempt_status in LOGIN_ATTEMPT_IN_FLIGHT_STATUSES:
             skipped.append(task.id)
             continue
-        task.status = "relogin_pending"
-        task.message = "等待重新登录，旧 Cookie 已保留"
-        task.finished_at = None
+        # 排队状态由 LoginBatchRunner 在批次启动时写入 attempt_*（含批次号）。
         task_ids.append(task.id)
 
     if not task_ids:

web/backend/routers/login.py

@@ -4,7 +4,7 @@ import asyncio
 import threading
 
 from fastapi import APIRouter, Depends, HTTPException, WebSocket, WebSocketDisconnect
-from sqlalchemy import func
+from sqlalchemy import func, or_
 from sqlalchemy.orm import Session, defer, load_only
 
 from ..database import SessionLocal, get_db
@@ -18,12 +18,24 @@ from ..services.login_service import LoginBatchRunner, batch_registry
 router = APIRouter(prefix="/api/login", tags=["登录任务"])
 
 
+def _effective_attempt_status():
+    """页面展示用的状态：优先最近一次登录尝试，没有尝试时回退到 CK 状态。
+
+    ``attempt_*`` 才是「这次登录怎么样了」，``status`` 只表示这条记录是否
+    持有可用 CK —— 两者分开，停止/失败/重启都不会让账号从列表里消失。
+    """
+    return func.coalesce(
+        func.nullif(LoginTask.attempt_status, ""),
+        LoginTask.status,
+    )
+
+
 def _login_task_summary(query):
     """按状态汇总登录任务，避免概览页拉完整任务列表。"""
     rows = (
         query.order_by(None)
-        .with_entities(LoginTask.status, func.count(LoginTask.id))
-        .group_by(LoginTask.status)
+        .with_entities(_effective_attempt_status(), func.count(LoginTask.id))
+        .group_by(_effective_attempt_status())
         .all()
     )
     status_counts = {status or "": count for status, count in rows}
@@ -120,7 +132,10 @@ def list_tasks(
         query = query.filter(Account.assigned_to == current.id)
 
     if batch_id:
-        query = query.filter(LoginTask.batch_id == batch_id)
+        # 记录持有的 CK 属于产出它的批次，最近一次尝试属于新批次，两者都要算。
+        query = query.filter(
+            or_(LoginTask.batch_id == batch_id, LoginTask.attempt_batch_id == batch_id)
+        )
 
     rows = query.order_by(LoginTask.id.desc()).limit(200).all()
     # 批量收集 account_id，一次性查出 username
@@ -137,18 +152,21 @@ def list_tasks(
 
     result = []
     for t in rows:
+        attempt_status = t.attempt_status or ""
         result.append(
             LoginTaskOut(
                 id=t.id,
-                batch_id=t.batch_id,
+                batch_id=t.attempt_batch_id or t.batch_id,
                 account_id=t.account_id,
                 account_username=accounts_map.get(t.account_id, ""),
-                status=t.status,
+                status=attempt_status or t.status,
                 cookie="",
-                message=t.message or "",
+                message=(t.attempt_message if attempt_status else "")
+                or t.message
+                or "",
                 created_by=t.created_by,
                 created_at=t.created_at,
-                finished_at=t.finished_at,
+                finished_at=t.attempt_at or t.finished_at,
             )
         )
     return result

web/backend/services/douyu_service.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 import uuid
 from datetime import UTC, datetime
 
-from sqlalchemy import and_, select
+from sqlalchemy import and_, func, select
 from sqlalchemy.orm import Session
 
 from core.douyu.activity_client import DouyuActivityClient
@@ -181,18 +181,16 @@ def ensure_douyu_config(db: Session) -> DouyuConfig:
     return config
 
 
-# 登录/重登批次正在处理的状态：这些记录已被批次接管，不允许并发操作。
-LOGIN_IN_FLIGHT_STATUSES = ("pending", "running", "relogin_pending", "relogin_running")
+# 登录/重登尝试的在途状态：判断只看 attempt_*，与记录持有的 CK 无关。
+LOGIN_ATTEMPT_IN_FLIGHT_STATUSES = ("pending", "running")
 
 
 def cookie_holding_filter():
     """判定登录记录当前是否仍持有 Cookie。
 
-    账号是否有 CK 只取决于记录里是否还留着 Cookie，而不是取决于批次状态：
-    普通登录在排队时会把复用的成功记录临时置为 ``pending``/``running``，
-    CK 重登失败会置为 ``relogin_failed``，这些状态都按设计保留旧 Cookie
-    （见 ``login_service`` 的「旧 Cookie 已保留」分支），账号必须继续可见。
-    只有 Cookie 真正被清空（账号检测成功）时才不再算持有 CK。
+    账号是否有 CK 只取决于记录里是否还留着 Cookie。登录尝试（attempt_*）
+    在成功前不会碰 status/cookie/message，所以停止、失败、进程重启都不会
+    让账号失去 CK；只有 Cookie 真正被清空（账号检测成功）时才不再持有。
     """
     return and_(
         LoginTask.cookie.isnot(None),
@@ -216,12 +214,14 @@ def latest_cookie_login_task(db: Session, account_id: int) -> LoginTask | None:
 def cookie_operable_filter():
     """判定登录记录能否基于现有 Cookie 再做检测或重登。
 
-    正在被登录/重登批次处理的行（pending/running/relogin_*）已由批次接管，
-    并发再点检测或重登只会互相覆盖状态，因此排除在外。
+    正在被批次处理的记录（attempt_* 处于在途状态）已由批次接管，并发再点
+    检测或重登只会互相覆盖，因此排除在外；CK 与 status 不参与判断。
     """
     return and_(
         cookie_holding_filter(),
-        LoginTask.status.notin_(LOGIN_IN_FLIGHT_STATUSES),
+        func.coalesce(LoginTask.attempt_status, "").notin_(
+            LOGIN_ATTEMPT_IN_FLIGHT_STATUSES
+        ),
     )
 
 

web/backend/services/login_service.py

@@ -11,6 +11,7 @@ from types import SimpleNamespace
 from typing import cast
 
 from loguru import logger
+from sqlalchemy import or_
 from sqlalchemy.orm import Session
 
 from core.douyu import DouyuLogin, IframeLoginAPI, WgapiLoginAPI
@@ -21,6 +22,7 @@ from ..models import Account as AccountModel
 from ..models import LoginTask
 from ..models import ProxyConfig as ProxyConfigModel
 from .cookie_check_service import check_douyu_cookie
+from .douyu_service import cookie_holding_filter
 
 
 def _create_api_strategy(strategy_name: str):
@@ -89,6 +91,66 @@ def _snapshot_proxy_config(
     )
 
 
+ATTEMPT_IN_FLIGHT_STATUSES = ("pending", "running")
+# 升级前把尝试状态写进 status 的历史取值；判断与清理时一并兼容。
+_LEGACY_IN_FLIGHT_STATUSES = (
+    "pending",
+    "running",
+    "relogin_pending",
+    "relogin_running",
+)
+
+
+def _attempt_in_flight(task: LoginTask) -> bool:
+    """判断这条记录是否正被某个批次处理（只看尝试字段，不看 CK 字段）。"""
+    if task.attempt_status in ATTEMPT_IN_FLIGHT_STATUSES:
+        return True
+    return not task.attempt_status and task.status in _LEGACY_IN_FLIGHT_STATUSES
+
+
+def _begin_attempt(task: LoginTask, batch_id: str, message: str) -> None:
+    """登记一次登录尝试。
+
+    只写 attempt_*：账号当前持有的 CK（status/cookie/message）在登录成功前
+    一律不动，所以批次被停止、失败或进程重启都不会让账号失去 CK。
+    """
+    task.attempt_status = "pending"
+    task.attempt_message = message
+    task.attempt_batch_id = batch_id
+    task.attempt_at = None
+
+
+def _mark_attempt_failed(
+    task: LoginTask, message: str, *, status: str = "failed"
+) -> None:
+    """记录一次失败/被中断的尝试；已经持有 CK 的记录保持原样。"""
+    completed_at = datetime.now(UTC)
+    task.attempt_status = status if status in {"failed", "error"} else "failed"
+    task.attempt_message = message
+    task.attempt_at = completed_at
+    if not task.cookie:
+        # 还没有任何 CK 需要保护：保持旧的失败语义，便于复用记录与页面统计。
+        task.status = task.attempt_status
+        task.message = message
+        task.finished_at = completed_at
+
+
+def _write_login_success(
+    task: LoginTask, *, cookie: str, message: str, batch_id: str
+) -> None:
+    """登录成功后把新 CK 写回记录：这是唯一替换旧 CK 的入口。"""
+    completed_at = datetime.now(UTC)
+    task.status = "success"
+    task.cookie = cookie
+    task.message = message
+    task.batch_id = batch_id
+    task.finished_at = completed_at
+    task.attempt_status = "success"
+    task.attempt_message = message
+    task.attempt_batch_id = batch_id
+    task.attempt_at = completed_at
+
+
 class LoginBatchRunner:
     """批量登录执行器，在线程中运行，通过 ThreadPoolExecutor 并发登录多个账号。"""
 
@@ -175,14 +237,12 @@ class LoginBatchRunner:
         task_id: int,
         message: str = "重新登录已停止，旧 Cookie 已保留",
     ) -> None:
-        """将未开始或已中断的重登任务收敛为保留旧 CK 的终态。"""
+        """把未开始或已中断的重登尝试收敛为终态，旧 CK 原封不动。"""
         worker_db = SessionLocal()
         try:
             task = worker_db.query(LoginTask).filter(LoginTask.id == task_id).first()
-            if task and task.status in ("relogin_pending", "relogin_running"):
-                task.status = "relogin_failed"
-                task.message = message
-                task.finished_at = datetime.now(UTC)
+            if task and _attempt_in_flight(task):
+                _mark_attempt_failed(task, message)
                 worker_db.commit()
         finally:
             worker_db.close()
@@ -219,8 +279,7 @@ class LoginBatchRunner:
         """在独立线程中执行单个账号登录，使用独立的 DB 会话。"""
         if self._stop.is_set():
             self._push_log("warning", f"任务已停止，跳过: {acc_info['username']}")
-            if self.mode == "relogin":
-                self._mark_relogin_stopped(task_id)
+            self._mark_relogin_stopped(task_id, "任务已停止，旧 Cookie 已保留")
             return
 
         worker_db = SessionLocal()
@@ -229,11 +288,15 @@ class LoginBatchRunner:
             if not task:
                 return
 
-            if self.mode == "relogin":
-                task.status = "relogin_running"
-                task.message = "正在重新登录，旧 Cookie 保留中"
-            else:
-                task.status = "running"
+            # 只推进尝试状态；记录持有的旧 CK 在成功前保持不变。
+            task.attempt_status = "running"
+            task.attempt_message = (
+                "正在重新登录，旧 Cookie 保留中"
+                if self.mode == "relogin"
+                else "正在检测"
+                if self.mode == "check"
+                else "正在登录"
+            )
             worker_db.commit()
 
             with self._counter_lock:
@@ -266,14 +329,14 @@ class LoginBatchRunner:
                     and not proxy_dict
                 ):
                     if self.mode == "relogin":
-                        task.status = "relogin_failed"
-                        task.message = (
-                            "重新登录失败: 代理不可用: 未配置代理（旧 Cookie 已保留）"
+                        _mark_attempt_failed(
+                            task,
+                            "重新登录失败: 代理不可用: 未配置代理（旧 Cookie 已保留）",
                         )
                     else:
-                        task.status = "error"
-                        task.message = "代理不可用: 未配置代理"
-                    task.finished_at = datetime.now(UTC)
+                        _mark_attempt_failed(
+                            task, "代理不可用: 未配置代理", status="error"
+                        )
                     worker_db.commit()
                     self._push_log(
                         "error", f"[{current}] {acc_info['username']} 代理不可用"
@@ -315,15 +378,18 @@ class LoginBatchRunner:
                     task.status = status
                     task.cookie = ""
                     task.message = result.message or CHECK_STATUS_MESSAGES[status]
+                    # 检测结果本身就是这次尝试的结论，页面按 attempt_* 展示。
+                    task.attempt_status = status
+                    task.attempt_message = task.message
+                    task.attempt_at = datetime.now(UTC)
+                    task.finished_at = datetime.now(UTC)
                     level = CHECK_STATUS_LOG_LEVELS.get(status, "info")
                     self._push_log(
                         level,
                         f"[{current}] {acc_info['username']} 检测结果: {task.message}",
                     )
                 elif result.success:
-                    task.status = "success"
-                    task.cookie = result.cookie
-                    task.message = result.message or "登录成功"
+                    message = result.message or "登录成功"
                     if self.mode == "relogin":
                         check_result = check_douyu_cookie(result.cookie)
                         task.ck_check_status = (
@@ -337,16 +403,26 @@ class LoginBatchRunner:
                         }
                         task.ck_checked_at = check_result["checked_at"]
                         if check_result["valid"]:
-                            task.message = "重新登录成功，Cookie 有效"
+                            message = "重新登录成功，Cookie 有效"
+                        else:
+                            message = f"重新登录成功，但 Cookie 有效性检测失败: {check_result['message']}"
+                    # 唯一替换旧 CK 的入口：登录成功后写回新 CK。
+                    _write_login_success(
+                        task,
+                        cookie=result.cookie,
+                        message=message,
+                        batch_id=self.batch_id,
+                    )
+                    if self.mode == "relogin":
+                        if task.ck_check_status == "valid":
                             self._push_log(
                                 "success",
                                 f"[{current}] {acc_info['username']} 重新登录成功，Cookie 已替换并验证有效",
                             )
                         else:
-                            task.message = f"重新登录成功，但 Cookie 有效性检测失败: {check_result['message']}"
                             self._push_log(
                                 "warning",
-                                f"[{current}] {acc_info['username']} 重新登录成功，但 Cookie 有效性检测失败: {check_result['message']}",
+                                f"[{current}] {acc_info['username']} 重新登录成功，但 Cookie 有效性检测失败: {task.ck_check_result.get('message') if task.ck_check_result else ''}",
                             )
                     else:
                         self._push_log(
@@ -355,18 +431,16 @@ class LoginBatchRunner:
                         )
                 else:
                     if self.mode == "relogin":
-                        # 重新登录失败时保留旧 Cookie 与成功状态，仅记录失败原因，行不消失
-                        task.status = "relogin_failed"
-                        task.message = (
-                            f"重新登录失败: {result.message}（旧 Cookie 已保留）"
+                        # 重登失败不动旧 CK 与成功状态，只记录这次尝试失败
+                        _mark_attempt_failed(
+                            task, f"重新登录失败: {result.message}（旧 Cookie 已保留）"
                         )
                         self._push_log(
                             "error",
                             f"[{current}] {acc_info['username']} 重新登录失败: {result.message}",
                         )
                     else:
-                        task.status = "failed"
-                        task.message = result.message
+                        _mark_attempt_failed(task, result.message or "登录失败")
                         self._push_log(
                             "error",
                             f"[{current}] {acc_info['username']} {action_name}失败: {result.message}",
@@ -374,20 +448,17 @@ class LoginBatchRunner:
 
             except Exception as e:  # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
                 if self.mode == "relogin":
-                    task.status = "relogin_failed"
-                    task.message = f"重新登录异常: {e}（旧 Cookie 已保留）"
+                    _mark_attempt_failed(task, f"重新登录异常: {e}（旧 Cookie 已保留）")
                     self._push_log(
                         "error", f"[{current}] {acc_info['username']} 重新登录异常: {e}"
                     )
                 else:
-                    task.status = "error"
-                    task.message = str(e)
+                    _mark_attempt_failed(task, str(e), status="error")
                     self._push_log(
                         "error",
                         f"[{current}] {acc_info['username']} {action_name}异常: {e}",
                     )
 
-            task.finished_at = datetime.now(UTC)
             worker_db.commit()
 
         finally:
@@ -416,15 +487,17 @@ class LoginBatchRunner:
                 self._push_log("info" if ok else "warning", f"白名单预热: {msg}")
 
         def _append_task_info(task: LoginTask, acc: AccountModel):
-            task.batch_id = batch_id
-            if self.mode == "relogin":
-                # 旧 Cookie 仅在 result.success 后由 _execute_one 原子替换。
-                task.status = "relogin_pending"
-                task.message = "等待重新登录，旧 Cookie 已保留"
-            else:
-                task.status = "pending"
-                task.message = ""
-            task.finished_at = None
+            # 只登记这次尝试：账号当前持有的 CK（status/cookie/message）在
+            # 登录成功前一律不动，批次被停止或失败都不会让账号失去 CK。
+            _begin_attempt(
+                task,
+                batch_id,
+                "等待重新登录，旧 Cookie 已保留"
+                if self.mode == "relogin"
+                else "等待检测"
+                if self.mode == "check"
+                else "等待登录",
+            )
             self.db.flush()
             task_infos.append(
                 {
@@ -491,50 +564,40 @@ class LoginBatchRunner:
                         self._push_log("warning", f"跳过无权账号: {acc.username}")
                         continue
 
-                    # 一个斗鱼账号只保留一条成功 CK：再次普通登录时更新最新成功记录。
-                    latest_success_task = (
+                    # 一个斗鱼账号只保留一条持有 CK 的记录：优先复用它，
+                    # 登录成功后才把新 CK 写回同一条记录。
+                    task = (
                         self.db.query(LoginTask)
                         .filter(
-                            LoginTask.account_id == aid, LoginTask.status == "success"
+                            LoginTask.account_id == aid,
+                            cookie_holding_filter(),
                         )
-                        .order_by(LoginTask.finished_at.desc(), LoginTask.id.desc())
+                        .order_by(LoginTask.id.desc())
                         .first()
                     )
-                    if latest_success_task:
-                        duplicate_success_tasks = (
-                            self.db.query(LoginTask)
-                            .filter(
-                                LoginTask.account_id == aid,
-                                LoginTask.status == "success",
-                                LoginTask.id != latest_success_task.id,
-                            )
-                            .all()
-                        )
-                        for duplicate_task in duplicate_success_tasks:
-                            self.db.delete(duplicate_task)
-                        task = latest_success_task
-                    else:
-                        # 复用该账号最近一条失败任务记录，避免重复产生多条失败历史。
-                        existing_task = (
+                    if task is None:
+                        # 复用最近一条失败记录，避免重复产生多条失败历史。
+                        task = (
                             self.db.query(LoginTask)
                             .filter(
                                 LoginTask.account_id == aid,
                                 LoginTask.status.in_(["failed", "error"]),
+                                or_(
+                                    LoginTask.cookie.is_(None),
+                                    LoginTask.cookie == "",
+                                ),
                             )
                             .order_by(LoginTask.id.desc())
                             .first()
                         )
-                        if existing_task:
-                            existing_task.cookie = ""
-                            task = existing_task
-                        else:
-                            task = LoginTask(
-                                batch_id=batch_id,
-                                account_id=aid,
-                                status="pending",
-                                created_by=self.created_by,
-                            )
-                            self.db.add(task)
+                    if task is None:
+                        task = LoginTask(
+                            batch_id=batch_id,
+                            account_id=aid,
+                            status="pending",
+                            created_by=self.created_by,
+                        )
+                        self.db.add(task)
 
                     _append_task_info(task, acc)
 
@@ -551,9 +614,12 @@ class LoginBatchRunner:
                 for item_index, item in enumerate(task_infos):
                     if self._stop.is_set():
                         self._push_log("warning", "任务已停止，跳过剩余账号")
-                        if self.mode == "relogin":
-                            for pending_item in task_infos[item_index:]:
-                                self._mark_relogin_stopped(pending_item["task_id"])
+                        # 所有模式都要把没轮到的尝试收敛掉，否则会一直停在
+                        # 「等待中」；已有 CK 的记录本身不受影响。
+                        for pending_item in task_infos[item_index:]:
+                            self._mark_relogin_stopped(
+                                pending_item["task_id"], "任务已停止，旧 Cookie 已保留"
+                            )
                         break
                     future = executor.submit(
                         self._execute_one,
@@ -624,51 +690,50 @@ class BatchRegistry:
 batch_registry = BatchRegistry()
 
 
+def _converge_interrupted_attempts(db: Session, message: str) -> int:
+    """把遗留的在途尝试收敛为失败。
+
+    只写 attempt_*；对已经没有 CK 的记录才顺带落一个失败终态。返回收敛条数。
+    """
+    tasks = (
+        db.query(LoginTask)
+        .filter(
+            or_(
+                LoginTask.attempt_status.in_(ATTEMPT_IN_FLIGHT_STATUSES),
+                LoginTask.status.in_(_LEGACY_IN_FLIGHT_STATUSES),
+            )
+        )
+        .all()
+    )
+    cleaned = 0
+    for task in tasks:
+        if not _attempt_in_flight(task):
+            continue
+        _mark_attempt_failed(task, message)
+        cleaned += 1
+    if cleaned:
+        db.commit()
+    return cleaned
+
+
 def cleanup_orphan_relogin_tasks(
     db: Session,
     message: str = "重新登录已中断（服务重启），旧 Cookie 已保留",
 ) -> int:
-    """服务重启后收敛遗留重登状态，避免页面永久显示重登中。"""
-    tasks = (
-        db.query(LoginTask)
-        .filter(LoginTask.status.in_(("relogin_pending", "relogin_running")))
-        .all()
-    )
-    if not tasks:
-        return 0
-
-    finished_at = datetime.now(UTC)
-    for task in tasks:
-        task.status = "relogin_failed"
-        task.message = message
-        task.finished_at = finished_at
-    db.commit()
-    return len(tasks)
+    """服务重启后收敛遗留的重登尝试，旧 CK 原封不动。"""
+    return _converge_interrupted_attempts(db, message)
 
 
 def cleanup_orphan_login_tasks(
     db: Session,
     message: str = "登录已中断（服务重启），旧 Cookie 已保留",
 ) -> int:
-    """服务重启后收敛遗留的普通登录状态。
+    """服务重启后收敛遗留的登录尝试。
 
-    普通登录批次停止或进程重启时，未轮到的记录会永远停在 pending/running，
-    页面上一直显示「排队中」，旧 CK 也失去终态说明。统一收敛为 failed 并
-    保留 Cookie，账号仍然可见可重登。
+    在途尝试只体现在 attempt_* 上，收敛它不会碰到账号已有的 CK；只有本来
+    就没有 CK 的记录才会顺带标记失败，避免页面永久显示「排队中」。
     """
-    tasks = (
-        db.query(LoginTask).filter(LoginTask.status.in_(("pending", "running"))).all()
-    )
-    if not tasks:
-        return 0
-
-    finished_at = datetime.now(UTC)
-    for task in tasks:
-        task.status = "failed"
-        task.message = message
-        task.finished_at = finished_at
-    db.commit()
-    return len(tasks)
+    return _converge_interrupted_attempts(db, message)
 
 
 # 在模块末尾导入 SessionLocal（避免循环导入）