feat: 异步沉淀主题摘要和成长档案

ai_knowledge_base_v2/apps/backend/app/api/admin_users.py

@@ -349,6 +349,34 @@ def generate_user_report(
     return api_success(periodic_report_dict(report), message="报告任务已进入后台队列")
 
 
+@router.post("/user/{user_id}/topic/{topic_id}/settlement/retry")
+def retry_topic_settlement(
+    user_id: int,
+    topic_id: int,
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_current_admin),
+) -> dict:
+    user = _get_user(db, user_id)
+    topic = db.scalar(
+        select(TopicSession).where(TopicSession.id == topic_id, TopicSession.user_id == user.id)
+    )
+    if topic is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="主题不存在")
+    if topic.status != "completed":
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="只能重试已结束主题的沉淀任务")
+    summary = GrowthProfileService.retry_topic_settlement(db, user=user, topic=topic)
+    OperationLogService.write(
+        db,
+        admin_id=current_admin.id,
+        module="topic_settlement",
+        action="retry",
+        target_id=summary.id,
+    )
+    db.commit()
+    db.refresh(summary)
+    return api_success(topic_summary_dict(summary), message="主题沉淀任务已重新进入队列")
+
+
 @router.put("/user/{user_id}")
 def update_user(
     user_id: int,

ai_knowledge_base_v2/apps/backend/app/api/chat.py

@@ -98,6 +98,15 @@ def finish_topic(
     return api_success(GrowthProfileService.finish_active_topic(db, user=current_user, session=session))
 
 
+@router.get("/topic/settlement/{summary_id}")
+def topic_settlement(
+    summary_id: int,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+) -> dict:
+    return api_success(GrowthProfileService.topic_settlement_result(db, user=current_user, summary_id=summary_id))
+
+
 @router.post("/session/{session_id}/help-card")
 def generate_help_card(
     session_id: int,

ai_knowledge_base_v2/apps/backend/app/api/user.py

@@ -48,6 +48,13 @@ def growth_profile(
                     "summary": item.summary,
                     "recommendedHomework": item.recommended_homework,
                     "nextObservation": item.next_observation,
+                    "status": item.status,
+                    "errorMessage": item.error_message,
+                    "attemptCount": item.attempt_count,
+                    "maxAttempts": item.max_attempts,
+                    "nextRunAt": item.next_run_at,
+                    "lastStartedAt": item.last_started_at,
+                    "finishedAt": item.finished_at,
                     "generatedAt": item.generated_at,
                 }
                 for item in summaries

ai_knowledge_base_v2/apps/backend/app/core/config.py

@@ -70,6 +70,10 @@ class Settings(BaseSettings):
     periodic_report_weekly_enabled: bool = True
     periodic_report_monthly_enabled: bool = True
     periodic_report_timezone: str = "Asia/Shanghai"
+    topic_settlement_worker_enabled: bool = True
+    topic_settlement_poll_seconds: int = 2
+    topic_settlement_stale_minutes: int = 30
+    topic_settlement_max_attempts: int = 3
     bootstrap_admin_username: str = ""
     bootstrap_admin_password: str = ""
     bootstrap_admin_name: str = "系统管理员"

ai_knowledge_base_v2/apps/backend/app/main.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 from contextlib import asynccontextmanager
 
 from fastapi import FastAPI
@@ -13,7 +14,7 @@ from app.core.observability import RequestObservabilityMiddleware, configure_log
 from app.services.secret_service import SecretService
 from app.services.maintenance_service import MaintenanceService
 from app.services.periodic_report_worker import PeriodicReportWorker
-import asyncio
+from app.services.topic_settlement_worker import TopicSettlementWorker
 
 
 @asynccontextmanager
@@ -24,12 +25,13 @@ async def lifespan(app: FastAPI):
         create_tables()
     maintenance_task = asyncio.create_task(MaintenanceService.run_forever())
     periodic_report_task = asyncio.create_task(PeriodicReportWorker.run_forever())
+    topic_settlement_task = asyncio.create_task(TopicSettlementWorker.run_forever())
     try:
         yield
     finally:
-        for task in (maintenance_task, periodic_report_task):
+        for task in (maintenance_task, periodic_report_task, topic_settlement_task):
             task.cancel()
-        for task in (maintenance_task, periodic_report_task):
+        for task in (maintenance_task, periodic_report_task, topic_settlement_task):
             try:
                 await task
             except asyncio.CancelledError:

ai_knowledge_base_v2/apps/backend/app/models/growth.py

@@ -12,7 +12,11 @@ PRIMARY_KEY_TYPE = BigInteger().with_variant(Integer, "sqlite")
 
 class TopicSummary(Base):
     __tablename__ = "sys_topic_summary"
-    __table_args__ = (UniqueConstraint("topic_session_id", name="uq_sys_topic_summary_topic"),)
+    __table_args__ = (
+        UniqueConstraint("topic_session_id", name="uq_sys_topic_summary_topic"),
+        Index("ix_topic_summary_job_due", "status", "next_run_at", "id"),
+        Index("ix_topic_summary_job_stale", "status", "locked_at"),
+    )
 
     id: Mapped[int] = mapped_column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True)
     topic_session_id: Mapped[int] = mapped_column(ForeignKey("sys_topic_session.id"), index=True, nullable=False)
@@ -28,6 +32,13 @@ class TopicSummary(Base):
     model_name: Mapped[str | None] = mapped_column(String(100), nullable=True)
     status: Mapped[str] = mapped_column(String(20), default="success", index=True, nullable=False)
     error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
+    attempt_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
+    max_attempts: Mapped[int] = mapped_column(Integer, default=3, nullable=False)
+    next_run_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    locked_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    locked_by: Mapped[str | None] = mapped_column(String(120), nullable=True)
+    last_started_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    finished_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
     generated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), nullable=False)
     created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), nullable=False)
     updated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), onupdate=func.now(), nullable=False)

ai_knowledge_base_v2/apps/backend/app/services/growth_profile_service.py

@@ -9,48 +9,153 @@ from fastapi import HTTPException, status
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
+from app.core.config import get_settings
 from app.models.chat import ChatMessage, ChatSession, TopicSession
 from app.models.growth import GrowthProfileRevision, TopicSummary, UserGrowthProfile
 from app.models.user import User
 from app.services.entitlement_service import EntitlementService
 from app.services.external_errors import ExternalServiceError
 from app.services.tracked_generation_service import TrackedGenerationService
-from app.services.topic_session_service import TopicSessionService
 
 
 class GrowthProfileService:
     @staticmethod
     def finish_active_topic(db: Session, *, user: User, session: ChatSession, force: bool = False) -> dict:
-        topic = TopicSessionService.active_for_session(db, user=user, session=session)
+        topic = db.scalar(
+            select(TopicSession)
+            .where(
+                TopicSession.user_id == user.id,
+                TopicSession.chat_session_id == session.id,
+                TopicSession.status == "active",
+            )
+            .order_by(TopicSession.created_at.desc(), TopicSession.id.desc())
+            .with_for_update()
+            .limit(1)
+        )
         if topic is None:
-            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="当前会话没有进行中的主题")
-        summary = GrowthProfileService.generate_topic_summary(db, user=user, topic=topic, force=force)
+            # 重复点击或客户端超时重试时返回最近一次任务，不重复生成摘要。
+            topic = db.scalar(
+                select(TopicSession)
+                .where(
+                    TopicSession.user_id == user.id,
+                    TopicSession.chat_session_id == session.id,
+                    TopicSession.status == "completed",
+                )
+                .order_by(TopicSession.ended_at.desc(), TopicSession.id.desc())
+                .limit(1)
+            )
+            summary = (
+                db.scalar(select(TopicSummary).where(TopicSummary.topic_session_id == topic.id))
+                if topic is not None
+                else None
+            )
+            if topic is None or summary is None:
+                raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="当前会话没有进行中的主题")
+            entitlement = EntitlementService.active_entitlement(db, user)
+            return _settlement_result(db, topic=topic, summary=summary, user=user, growth_enabled=entitlement.enable_growth_profile)
+
+        has_messages = db.scalar(
+            select(ChatMessage.id)
+            .where(ChatMessage.topic_session_id == topic.id, ChatMessage.user_id == user.id)
+            .limit(1)
+        )
+        if has_messages is None:
+            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="当前主题还没有可沉淀的对话内容")
+
+        summary = db.scalar(select(TopicSummary).where(TopicSummary.topic_session_id == topic.id))
+        is_new_summary = summary is None
+        if is_new_summary:
+            summary = TopicSummary(topic_session_id=topic.id, user_id=user.id, summary="")
+        if is_new_summary or force or summary.status not in {"pending", "running", "success", "fallback"}:
+            _reset_summary_job(summary)
+        summary.max_attempts = max(1, get_settings().topic_settlement_max_attempts)
+        db.add(summary)
+
         topic.status = "completed"
         topic.ended_at = _now()
-        topic.recommended_homework = summary.recommended_homework
         db.add(topic)
 
         entitlement = EntitlementService.active_entitlement(db, user)
-        profile = None
-        if entitlement.enable_growth_profile:
-            profile = GrowthProfileService.update_growth_profile(db, user=user, topic_summary=summary)
         db.commit()
         db.refresh(topic)
         db.refresh(summary)
-        if profile is not None:
-            db.refresh(profile)
-        return {
-            "topic": topic_dict(topic),
-            "summary": topic_summary_dict(summary),
-            "profile": growth_profile_dict(profile) if profile is not None else None,
-            "growthProfileEnabled": entitlement.enable_growth_profile,
-        }
+        return _settlement_result(db, topic=topic, summary=summary, user=user, growth_enabled=entitlement.enable_growth_profile)
 
     @staticmethod
-    def generate_topic_summary(db: Session, *, user: User, topic: TopicSession, force: bool = False) -> TopicSummary:
+    def retry_topic_settlement(db: Session, *, user: User, topic: TopicSession) -> TopicSummary:
+        summary = db.scalar(select(TopicSummary).where(TopicSummary.topic_session_id == topic.id).with_for_update())
+        if summary is None:
+            summary = TopicSummary(topic_session_id=topic.id, user_id=user.id, summary="")
+        if summary.status == "running":
+            return summary
+        _reset_summary_job(summary)
+        summary.max_attempts = max(1, get_settings().topic_settlement_max_attempts)
+        db.add(summary)
+        db.flush()
+        return summary
+
+    @staticmethod
+    def topic_settlement_result(db: Session, *, user: User, summary_id: int) -> dict:
+        summary = db.scalar(
+            select(TopicSummary).where(TopicSummary.id == summary_id, TopicSummary.user_id == user.id)
+        )
+        if summary is None:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="主题沉淀任务不存在")
+        topic = db.get(TopicSession, summary.topic_session_id)
+        if topic is None:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="主题不存在")
+        entitlement = EntitlementService.active_entitlement(db, user)
+        return _settlement_result(
+            db,
+            topic=topic,
+            summary=summary,
+            user=user,
+            growth_enabled=entitlement.enable_growth_profile,
+        )
+
+    @staticmethod
+    def process_topic_settlement(db: Session, *, summary: TopicSummary) -> tuple[TopicSummary, UserGrowthProfile | None]:
+        topic = db.get(TopicSession, summary.topic_session_id)
+        # Redis 失效时允许多个进程依靠数据库继续消费；用户行锁保证同一学员的
+        # 多个主题不会并发覆盖长期成长档案。
+        user = db.scalar(select(User).where(User.id == summary.user_id).with_for_update())
+        if topic is None or user is None or user.is_deleted:
+            raise ValueError("主题或用户不存在")
+        generated = GrowthProfileService.generate_topic_summary(
+            db,
+            user=user,
+            topic=topic,
+            force=True,
+            process_job=True,
+            allow_fallback=False,
+        )
+        topic.recommended_homework = generated.recommended_homework
+        db.add(topic)
+        entitlement = EntitlementService.active_entitlement(db, user)
+        profile = None
+        if entitlement.enable_growth_profile:
+            profile = GrowthProfileService.update_growth_profile(db, user=user, topic_summary=generated)
+        return generated, profile
+
+    @staticmethod
+    def generate_topic_summary(
+        db: Session,
+        *,
+        user: User,
+        topic: TopicSession,
+        force: bool = False,
+        process_job: bool = False,
+        allow_fallback: bool = True,
+    ) -> TopicSummary:
         existing = db.scalar(select(TopicSummary).where(TopicSummary.topic_session_id == topic.id))
         if existing is not None and not force:
-            return existing
+            if existing.status in {"success", "fallback"}:
+                return existing
+            if existing.status in {"pending", "running"}:
+                raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="主题正在后台沉淀，请稍后再试")
+            raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="主题沉淀失败，请稍后重试")
+        if existing is not None and existing.status in {"pending", "running"} and not process_job:
+            raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="主题正在后台沉淀，请稍后再试")
         messages = list(
             db.scalars(
                 select(ChatMessage)
@@ -77,6 +182,8 @@ class GrowthProfileService:
             status_value = "success"
             error_message = None
         except ExternalServiceError as exc:
+            if not allow_fallback:
+                raise
             data = _fallback_summary(topic, conversation, "")
             status_value = "fallback"
             error_message = str(exc)
@@ -88,6 +195,10 @@ class GrowthProfileService:
         existing.status = status_value
         existing.error_message = error_message
         existing.generated_at = _now()
+        existing.finished_at = _now()
+        existing.next_run_at = None
+        existing.locked_at = None
+        existing.locked_by = None
         db.add(existing)
         db.flush()
         return existing
@@ -95,6 +206,8 @@ class GrowthProfileService:
     @staticmethod
     def update_growth_profile(db: Session, *, user: User, topic_summary: TopicSummary) -> UserGrowthProfile:
         profile = db.scalar(select(UserGrowthProfile).where(UserGrowthProfile.user_id == user.id))
+        if profile is not None and profile.last_topic_summary_id == topic_summary.id:
+            return profile
         before = growth_profile_dict(profile) if profile is not None else None
         if profile is None:
             profile = UserGrowthProfile(user_id=user.id, profile_text="")
@@ -197,6 +310,11 @@ def topic_summary_dict(summary: TopicSummary) -> dict:
         "modelName": summary.model_name,
         "status": summary.status,
         "errorMessage": summary.error_message,
+        "attemptCount": summary.attempt_count,
+        "maxAttempts": summary.max_attempts,
+        "nextRunAt": summary.next_run_at,
+        "lastStartedAt": summary.last_started_at,
+        "finishedAt": summary.finished_at,
         "generatedAt": summary.generated_at,
     }
 
@@ -303,17 +421,32 @@ def _parse_summary_json(raw: str) -> dict[str, Any] | None:
     text = raw.strip()
     if not text:
         return None
-    if text.startswith("```"):
-        text = re.sub(r"^```(?:json)?", "", text, flags=re.IGNORECASE).strip()
-        text = re.sub(r"```$", "", text).strip()
-    match = re.search(r"\{[\s\S]*\}", text)
-    if match:
-        text = match.group(0)
-    try:
-        parsed = json.loads(text)
-    except json.JSONDecodeError:
-        return None
-    return parsed if isinstance(parsed, dict) else None
+    # 部分模型会把推理过程和最终 JSON 一起放进 answer，甚至在推理中先给出一份
+    # 草稿 JSON。先移除推理块，再从后向前选取可完整解析的对象，避免把思考过程
+    # 或多个 JSON 拼接后写入用户成长档案。
+    text = re.sub(r"<(?:think|analysis)>[\s\S]*?</(?:think|analysis)>", "", text, flags=re.IGNORECASE).strip()
+    fenced = re.findall(r"```(?:json)?\s*([\s\S]*?)\s*```", text, flags=re.IGNORECASE)
+    candidates = [*fenced, text]
+    for candidate in reversed(candidates):
+        parsed = _last_json_object(candidate)
+        if parsed is not None:
+            return parsed
+    return None
+
+
+def _last_json_object(text: str) -> dict[str, Any] | None:
+    decoder = json.JSONDecoder()
+    parsed_objects: list[dict[str, Any]] = []
+    for index, character in enumerate(text):
+        if character != "{":
+            continue
+        try:
+            value, _ = decoder.raw_decode(text[index:])
+        except json.JSONDecodeError:
+            continue
+        if isinstance(value, dict):
+            parsed_objects.append(value)
+    return parsed_objects[-1] if parsed_objects else None
 
 
 def _messages_text(messages: list[ChatMessage]) -> str:
@@ -341,3 +474,43 @@ def _limit(text: str, max_len: int) -> str:
 
 def _now() -> datetime:
     return datetime.now(UTC).replace(tzinfo=None)
+
+
+def _reset_summary_job(summary: TopicSummary) -> None:
+    summary.summary = ""
+    summary.main_events = None
+    summary.emotions = None
+    summary.body_feelings = None
+    summary.beliefs = None
+    summary.recommended_homework = None
+    summary.insights = None
+    summary.next_observation = None
+    summary.model_name = None
+    summary.status = "pending"
+    summary.error_message = None
+    summary.attempt_count = 0
+    summary.next_run_at = _now()
+    summary.locked_at = None
+    summary.locked_by = None
+    summary.last_started_at = None
+    summary.finished_at = None
+
+
+def _settlement_result(
+    db: Session,
+    *,
+    topic: TopicSession,
+    summary: TopicSummary,
+    user: User,
+    growth_enabled: bool,
+) -> dict:
+    profile = None
+    if growth_enabled and summary.status in {"success", "fallback"}:
+        profile = GrowthProfileService.get_growth_profile(db, user.id)
+    return {
+        "topic": topic_dict(topic),
+        "summary": topic_summary_dict(summary),
+        "profile": growth_profile_dict(profile),
+        "growthProfileEnabled": growth_enabled,
+        "settlementStatus": summary.status,
+    }

ai_knowledge_base_v2/apps/backend/app/services/topic_settlement_worker.py

@@ -0,0 +1,217 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+import socket
+import uuid
+from datetime import UTC, datetime, timedelta
+
+from sqlalchemy import or_, select
+from sqlalchemy.orm import Session
+
+from app.core.config import get_settings
+from app.core.database import SessionLocal
+from app.models.chat import TopicSession
+from app.models.growth import TopicSummary
+from app.models.user import User
+from app.services.growth_profile_service import GrowthProfileService
+from app.services.redis_client import get_sync_redis_client
+
+
+logger = logging.getLogger(__name__)
+WORKER_LOCK_KEY = "topic-settlement:worker:lock"
+WORKER_LOCK_TTL_SECONDS = 900
+
+
+class TopicSettlementWorker:
+    """持久化主题沉淀队列。
+
+    MySQL 保存任务、重试和执行状态，Redis 仅限制多进程同时执行数量。
+    即使 Redis 暂时不可用或服务重启，已结束主题也不会丢失沉淀任务。
+    """
+
+    @classmethod
+    async def run_forever(cls) -> None:
+        settings = get_settings()
+        if not settings.topic_settlement_worker_enabled:
+            logger.info("topic settlement worker disabled")
+            return
+        worker_id = f"{socket.gethostname()}:{os.getpid()}:{uuid.uuid4().hex[:8]}"
+        poll_seconds = max(1, settings.topic_settlement_poll_seconds)
+        while True:
+            try:
+                processed = await asyncio.to_thread(cls.run_once, worker_id)
+            except Exception:
+                processed = False
+                logger.exception("topic settlement worker iteration failed")
+            await asyncio.sleep(0 if processed else poll_seconds)
+
+    @classmethod
+    def run_once(cls, worker_id: str) -> bool:
+        redis, acquired = _acquire_worker_lock(worker_id)
+        if not acquired:
+            return False
+        try:
+            now = _now()
+            with SessionLocal() as db:
+                cls.recover_stale_jobs(db, now=now)
+                db.commit()
+            with SessionLocal() as db:
+                summary_id = cls.claim_next(db, worker_id=worker_id, now=now)
+            if summary_id is None:
+                return False
+            with SessionLocal() as db:
+                cls.execute_claimed(db, summary_id=summary_id, worker_id=worker_id)
+            return True
+        finally:
+            _release_worker_lock(redis, worker_id)
+
+    @staticmethod
+    def claim_next(db: Session, *, worker_id: str, now: datetime | None = None) -> int | None:
+        current = now or _now()
+        summary = db.scalar(
+            select(TopicSummary)
+            .where(
+                TopicSummary.status == "pending",
+                TopicSummary.attempt_count < TopicSummary.max_attempts,
+                or_(TopicSummary.next_run_at.is_(None), TopicSummary.next_run_at <= current),
+            )
+            .order_by(TopicSummary.next_run_at.asc(), TopicSummary.id.asc())
+            .with_for_update(skip_locked=True)
+            .limit(1)
+        )
+        if summary is None:
+            db.rollback()
+            return None
+        summary.status = "running"
+        summary.attempt_count += 1
+        summary.locked_at = current
+        summary.locked_by = worker_id
+        summary.last_started_at = current
+        summary.finished_at = None
+        db.add(summary)
+        db.commit()
+        return summary.id
+
+    @staticmethod
+    def execute_claimed(db: Session, *, summary_id: int, worker_id: str) -> TopicSummary | None:
+        summary = db.get(TopicSummary, summary_id)
+        if summary is None or summary.status != "running" or summary.locked_by != worker_id:
+            return summary
+        topic = db.get(TopicSession, summary.topic_session_id)
+        user = db.get(User, summary.user_id)
+        if topic is None or user is None or user.is_deleted:
+            _finish_permanent_failure(summary, "主题或用户不存在")
+            db.commit()
+            return summary
+
+        try:
+            GrowthProfileService.process_topic_settlement(db, summary=summary)
+            now = _now()
+            summary.next_run_at = None
+            summary.finished_at = now
+            summary.locked_at = None
+            summary.locked_by = None
+            db.add(summary)
+            db.commit()
+        except Exception as exc:
+            # 撤销本轮可能已经写入的半成品摘要、档案和修订记录，再单独更新任务状态。
+            db.rollback()
+            summary = db.get(TopicSummary, summary_id)
+            if summary is None:
+                return None
+            summary.error_message = str(exc)[:2000]
+            now = _now()
+            if summary.attempt_count < summary.max_attempts:
+                retry_seconds = min(300, 30 * (2 ** max(0, summary.attempt_count - 1)))
+                summary.status = "pending"
+                summary.next_run_at = now + timedelta(seconds=retry_seconds)
+                summary.finished_at = None
+            else:
+                summary.status = "failed"
+                summary.next_run_at = None
+                summary.finished_at = now
+            summary.locked_at = None
+            summary.locked_by = None
+            db.add(summary)
+            db.commit()
+        db.refresh(summary)
+        return summary
+
+    @staticmethod
+    def recover_stale_jobs(db: Session, *, now: datetime | None = None) -> int:
+        current = now or _now()
+        stale_before = current - timedelta(minutes=max(5, get_settings().topic_settlement_stale_minutes))
+        summaries = list(
+            db.scalars(
+                select(TopicSummary)
+                .where(
+                    TopicSummary.status == "running",
+                    TopicSummary.locked_at.is_not(None),
+                    TopicSummary.locked_at < stale_before,
+                )
+                .order_by(TopicSummary.id.asc())
+                .limit(100)
+                .with_for_update(skip_locked=True)
+            )
+        )
+        for summary in summaries:
+            if summary.attempt_count >= summary.max_attempts:
+                summary.status = "failed"
+                summary.finished_at = current
+                summary.next_run_at = None
+            else:
+                summary.status = "pending"
+                summary.next_run_at = current
+                summary.finished_at = None
+            summary.error_message = _append_error(summary.error_message, "任务执行进程中断，系统已自动恢复")
+            summary.locked_at = None
+            summary.locked_by = None
+            db.add(summary)
+        return len(summaries)
+
+
+def _finish_permanent_failure(summary: TopicSummary, message: str) -> None:
+    summary.status = "failed"
+    summary.error_message = message
+    summary.next_run_at = None
+    summary.locked_at = None
+    summary.locked_by = None
+    summary.finished_at = _now()
+
+
+def _append_error(current: str | None, message: str) -> str:
+    if not current:
+        return message
+    return f"{current}\n{message}"[-2000:]
+
+
+def _acquire_worker_lock(worker_id: str):
+    redis = get_sync_redis_client()
+    if redis is None:
+        return None, True
+    try:
+        return redis, bool(redis.set(WORKER_LOCK_KEY, worker_id, nx=True, ex=WORKER_LOCK_TTL_SECONDS))
+    except Exception:
+        logger.warning("redis unavailable for topic settlement lock; falling back to database claim", exc_info=True)
+        return None, True
+
+
+def _release_worker_lock(redis, worker_id: str) -> None:
+    if redis is None:
+        return
+    try:
+        redis.eval(
+            "if redis.call('get', KEYS[1]) == ARGV[1] then "
+            "return redis.call('del', KEYS[1]) else return 0 end",
+            1,
+            WORKER_LOCK_KEY,
+            worker_id,
+        )
+    except Exception:
+        logger.warning("failed to release topic settlement worker lock", exc_info=True)
+
+
+def _now() -> datetime:
+    return datetime.now(UTC).replace(tzinfo=None)