feat: permanently delete archived lists

backend/main.py

@@ -1,11 +1,14 @@
 import base64
 import json
+import logging
+import os
+import shutil
 import time
 from collections import defaultdict, deque
 from contextlib import asynccontextmanager
 from datetime import datetime, timedelta
-from pathlib import Path
-from uuid import UUID
+from pathlib import Path, PureWindowsPath
+from uuid import UUID, uuid4
 
 from fastapi import Depends, FastAPI, HTTPException, Query, Request, Response
 from fastapi.openapi.docs import get_swagger_ui_html
@@ -26,7 +29,19 @@ from .auth import (
     verify_password,
 )
 from .db import create_schema, get_db
-from .models import AppState, Folder, RecurrenceTemplate, Session, Task, TaskList, User, utcnow
+from .models import (
+    AppState,
+    Attachment,
+    Folder,
+    PurgeOperation,
+    RecurrenceException,
+    RecurrenceTemplate,
+    Session,
+    Task,
+    TaskList,
+    User,
+    utcnow,
+)
 from .mvp import audit, occurrences
 from .mvp import router as mvp_router
 from .schemas import (
@@ -99,6 +114,7 @@ async def openapi(_: User = Depends(current_user)):
 
 
 app.include_router(mvp_router)
+logger = logging.getLogger(__name__)
 _login_attempts: dict[tuple[str, str], deque[float]] = defaultdict(deque)
 
 
@@ -569,6 +585,237 @@ async def restore_list(
     return item
 
 
+def _attachment_path(attachment_root: Path, storage_name: str) -> Path:
+    if not storage_name or "\x00" in storage_name:
+        raise ValueError(storage_name)
+    windows_path = PureWindowsPath(storage_name)
+    if windows_path.is_absolute() or windows_path.drive:
+        raise ValueError(storage_name)
+    parts = storage_name.replace("\\", "/").split("/")
+    if any(part in {"", ".", ".."} for part in parts):
+        raise ValueError(storage_name)
+    relative_path = Path(*parts)
+    resolved_path = (attachment_root / relative_path).resolve()
+    try:
+        resolved_path.relative_to(attachment_root)
+    except ValueError as exc:
+        raise ValueError(storage_name) from exc
+    return resolved_path
+
+
+def _create_unique_quarantine(attachment_root: Path) -> tuple[UUID, Path]:
+    while True:
+        operation_id = uuid4()
+        trash_dir = attachment_root / ".purge-trash" / str(operation_id)
+        try:
+            trash_dir.mkdir(parents=True, exist_ok=False)
+        except FileExistsError:
+            continue
+        return operation_id, trash_dir
+
+
+def _remove_owned_quarantine(trash_dir: Path) -> None:
+    shutil.rmtree(trash_dir)
+    try:
+        trash_dir.parent.rmdir()
+    except OSError:
+        pass
+
+
+def _restore_quarantined_files(
+    moved_files: list[tuple[Path, Path]], *, list_id: UUID, operation_id: UUID
+) -> list[str]:
+    restore_errors: list[str] = []
+    for source_path, trash_path in reversed(moved_files):
+        try:
+            source_path.parent.mkdir(parents=True, exist_ok=True)
+            os.replace(trash_path, source_path)
+        except OSError:
+            restore_errors.append(str(trash_path))
+            logger.exception(
+                "List purge attachment restore failed",
+                extra={
+                    "list_id": str(list_id),
+                    "operation_id": str(operation_id),
+                    "trash_path": str(trash_path),
+                },
+            )
+    return restore_errors
+
+
+async def _retry_purge_cleanup(db: AsyncSession, operation: PurgeOperation) -> Response:
+    trash_dir = Path(operation.trash_dir)
+    try:
+        if trash_dir.exists():
+            _remove_owned_quarantine(trash_dir)
+    except OSError as exc:
+        logger.exception(
+            "List purge retry cleanup failed: %s",
+            trash_dir,
+            extra={
+                "list_id": str(operation.list_id),
+                "operation_id": str(operation.id),
+                "trash_path": str(trash_dir),
+            },
+        )
+        raise HTTPException(
+            status_code=500,
+            detail=(
+                "清单数据已删除，但附件清理未完成，"
+                f"操作ID: {operation.id}, 隔离路径: {trash_dir}"
+            ),
+        ) from exc
+    await db.delete(operation)
+    await db.commit()
+    return Response(status_code=204)
+
+
+@app.delete("/api/v1/lists/{list_id}/purge", status_code=204)
+async def purge_list(
+    list_id: UUID,
+    user: User = Depends(current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    from .config import get_settings
+
+    item = await db.scalar(
+        select(TaskList)
+        .where(TaskList.id == list_id, TaskList.user_id == user.id)
+        .with_for_update()
+    )
+    if item is None:
+        pending_operation = await db.scalar(
+            select(PurgeOperation)
+            .where(PurgeOperation.list_id == list_id, PurgeOperation.user_id == user.id)
+            .with_for_update()
+        )
+        if pending_operation is not None:
+            return await _retry_purge_cleanup(db, pending_operation)
+        raise HTTPException(status_code=404, detail="清单不存在")
+    if item.is_inbox:
+        raise HTTPException(status_code=409, detail="系统收集箱不能永久删除")
+    if item.deleted_at is None:
+        raise HTTPException(status_code=409, detail="请先归档再永久删除")
+
+    task_ids = select(Task.id).where(Task.list_id == item.id)
+    template_ids = select(RecurrenceTemplate.id).where(RecurrenceTemplate.task_id.in_(task_ids))
+    storage_names = list(
+        (await db.scalars(select(Attachment.storage_name).where(Attachment.task_id.in_(task_ids)))).all()
+    )
+    attachment_root = Path(get_settings().attachment_dir).resolve()
+    try:
+        attachment_paths = [_attachment_path(attachment_root, name) for name in storage_names]
+    except ValueError as exc:
+        await db.rollback()
+        logger.error(
+            "List purge rejected unsafe attachment path",
+            extra={"list_id": str(list_id), "storage_name": str(exc)},
+        )
+        raise HTTPException(
+            status_code=409, detail="附件存储路径无效，无法永久删除清单"
+        ) from exc
+
+    moved_files: list[tuple[Path, Path]] = []
+    owns_trash_dir = False
+    operation_id = uuid4()
+    trash_dir = attachment_root / ".purge-trash" / str(operation_id)
+    try:
+        operation_id, trash_dir = _create_unique_quarantine(attachment_root)
+        owns_trash_dir = True
+        for source_path in attachment_paths:
+            if not source_path.is_file():
+                continue
+            trash_path = trash_dir / source_path.relative_to(attachment_root)
+            trash_path.parent.mkdir(parents=True, exist_ok=True)
+            os.replace(source_path, trash_path)
+            moved_files.append((source_path, trash_path))
+    except OSError as exc:
+        restore_errors = _restore_quarantined_files(
+            moved_files, list_id=list_id, operation_id=operation_id
+        )
+        if owns_trash_dir and not restore_errors:
+            _remove_owned_quarantine(trash_dir)
+        await db.rollback()
+        if restore_errors:
+            raise HTTPException(
+                status_code=500,
+                detail=(
+                    "附件隔离失败且部分附件恢复失败，"
+                    f"操作ID: {operation_id}, 隔离路径: {trash_dir}, "
+                    f"未恢复: {', '.join(restore_errors)}"
+                ),
+            ) from exc
+        raise HTTPException(status_code=500, detail="附件隔离失败，清单未删除") from exc
+
+    operation = PurgeOperation(
+        id=operation_id,
+        list_id=list_id,
+        user_id=user.id,
+        trash_dir=str(trash_dir),
+        status="cleanup_pending",
+    )
+    try:
+        await db.execute(
+            delete(RecurrenceException).where(RecurrenceException.template_id.in_(template_ids))
+        )
+        await db.execute(delete(RecurrenceTemplate).where(RecurrenceTemplate.task_id.in_(task_ids)))
+        await db.execute(delete(Attachment).where(Attachment.task_id.in_(task_ids)))
+        await db.execute(delete(Task).where(Task.list_id == item.id))
+        await db.delete(item)
+        db.add(operation)
+        await db.commit()
+    except Exception as exc:
+        await db.rollback()
+        restore_errors = _restore_quarantined_files(
+            moved_files, list_id=list_id, operation_id=operation_id
+        )
+        if not restore_errors:
+            _remove_owned_quarantine(trash_dir)
+            raise HTTPException(status_code=500, detail="数据库提交失败，清单未删除") from exc
+        raise HTTPException(
+            status_code=500,
+            detail=(
+                "数据库提交失败且部分附件恢复失败，"
+                f"操作ID: {operation_id}, 隔离路径: {trash_dir}, "
+                f"未恢复: {', '.join(restore_errors)}"
+            ),
+        ) from exc
+
+    try:
+        _remove_owned_quarantine(trash_dir)
+    except OSError as exc:
+        logger.exception(
+            "List purge committed but quarantine cleanup failed: %s",
+            trash_dir,
+            extra={
+                "list_id": str(list_id),
+                "operation_id": str(operation_id),
+                "trash_path": str(trash_dir),
+            },
+        )
+        raise HTTPException(
+            status_code=500,
+            detail=(
+                "清单数据已删除，但附件清理未完成，"
+                f"操作ID: {operation_id}, 隔离路径: {trash_dir}"
+            ),
+        ) from exc
+
+    await db.delete(operation)
+    try:
+        await db.commit()
+    except Exception as exc:
+        await db.rollback()
+        raise HTTPException(
+            status_code=500,
+            detail=(
+                "清单数据及附件已删除，但清理操作状态未完成，"
+                f"操作ID: {operation_id}；请重试永久删除"
+            ),
+        ) from exc
+    return Response(status_code=204)
+
+
 
 def _encode_cursor(position: int, created_at: datetime, task_id: UUID) -> str:
     raw = json.dumps([position, created_at.isoformat(), str(task_id)]).encode()

backend/models.py

@@ -78,6 +78,16 @@ class TaskList(Base):
 
 
 
+class PurgeOperation(Base):
+    __tablename__ = "purge_operations"
+    id: Mapped[UUID] = mapped_column(primary_key=True, default=new_id)
+    list_id: Mapped[UUID] = mapped_column(unique=True, index=True)
+    user_id: Mapped[UUID] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True)
+    trash_dir: Mapped[str] = mapped_column(String(1024))
+    status: Mapped[str] = mapped_column(String(32), default="cleanup_pending")
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
+
+
 class Task(Base):
     __tablename__ = "tasks"
     __table_args__ = (

frontend/src/App.vue

@@ -10,6 +10,7 @@ import { formatApiErrorDetail, isTaskView, nextTotalAfterLocalTaskAdd, normalize
 import { csrfHeader } from './lib/csrf'
 import { createCompletionPulse } from './lib/completion-motion'
 import { captureListDragPointer, getAdjacentListMove, hasExceededLongPressMovement, moveListToScope, snapshotListDragPointer, type ListDragPointer } from './lib/list-drag'
+import { nextDialogFocusIndex } from './lib/list-purge'
 import MvpPanel from './MvpPanel.vue'
 import CountdownPanel from './CountdownPanel.vue'
 import FloatingAddButton from './components/FloatingAddButton.vue'
@@ -30,6 +31,12 @@ const password = ref('')
 const folders = ref<FolderItem[]>([])
 const lists = ref<TaskList[]>([])
 const archivedLists = ref<TaskList[]>([])
+const purgeListTarget = ref<TaskList | null>(null)
+const purgeListSubmitting = ref(false)
+const purgeListError = ref('')
+const purgeCancelButton = ref<HTMLButtonElement | null>(null)
+const purgeListDialog = ref<HTMLElement | null>(null)
+let purgeListTrigger: HTMLElement | null = null
 const tasks = ref<Task[]>([])
 const overdueTasks = ref<Task[]>([])
 const trash = ref<Task[]>([])
@@ -779,7 +786,19 @@ async function deleteEntity(kind: 'folders' | 'lists', item: FolderItem | TaskLi
   if (kind === 'lists') {
     const answer = await askText(`归档清单「${item.name}」？`, '', '', '归档')
     if (answer === null) return
-    try { await api(`/${kind}/${item.id}`, { method: 'DELETE' }); await loadArchivedLists(); await refreshAll(); toast('清单已归档') } catch (reason) { fail(reason) }
+    try {
+      const wasCurrentList = activeView.value === 'tasks' && activeList.value === item.id
+      await api(`/${kind}/${item.id}`, { method: 'DELETE' })
+      await loadArchivedLists()
+      await refreshAll()
+      if (wasCurrentList) {
+        selectedTask.value = null
+        mobileDetail.value = false
+        const inboxId = lists.value.find((list) => list.is_inbox)?.id || ''
+        await switchView('tasks', inboxId)
+      }
+      toast('清单已归档')
+    } catch (reason) { fail(reason) }
     return
   }
   const answer = await askText(`删除文件夹「${item.name}」？`, '', '', '删除')
@@ -792,6 +811,43 @@ async function loadArchivedLists() {
 async function restoreList(item: TaskList) {
   try { await api(`/lists/${item.id}/restore`, { method: 'POST' }); await loadArchivedLists(); await refreshAll(); toast('清单已恢复') } catch (reason) { fail(reason) }
 }
+function openPurgeList(item: TaskList, trigger?: EventTarget | null) {
+  purgeListTarget.value = item
+  purgeListError.value = ''
+  purgeListTrigger = trigger instanceof HTMLElement ? trigger : document.activeElement as HTMLElement | null
+  nextTick(() => purgeCancelButton.value?.focus())
+}
+function closePurgeList() {
+  if (purgeListSubmitting.value) return
+  purgeListTarget.value = null
+  purgeListError.value = ''
+  nextTick(() => purgeListTrigger?.focus())
+}
+function handlePurgeDialogKeydown(event: KeyboardEvent) {
+  if (event.key === 'Escape' && !purgeListSubmitting.value) closePurgeList()
+  if (event.key !== 'Tab' || !purgeListDialog.value) return
+  const controls = [...purgeListDialog.value.querySelectorAll<HTMLElement>('button:not(:disabled)')]
+  if (!controls.length) return
+  const activeIndex = controls.indexOf(document.activeElement as HTMLElement)
+  const nextIndex = nextDialogFocusIndex(activeIndex, controls.length, event.shiftKey)
+  if (nextIndex !== null) { event.preventDefault(); controls[nextIndex].focus() }
+}
+async function confirmPurgeList() {
+  if (!purgeListTarget.value || purgeListSubmitting.value) return
+  purgeListSubmitting.value = true
+  purgeListError.value = ''
+  try {
+    const purgedId = purgeListTarget.value.id
+    await api(`/lists/${purgeListTarget.value.id}/purge`, { method: 'DELETE' })
+    archivedLists.value = archivedLists.value.filter((list) => list.id !== purgedId)
+    purgeListTarget.value = null
+    toast('清单已永久删除')
+  } catch (reason) {
+    purgeListError.value = reason instanceof Error ? reason.message : '永久删除失败'
+  } finally {
+    purgeListSubmitting.value = false
+  }
+}
 function toggleSidebarCreate() { sidebarCreateOpen.value = !sidebarCreateOpen.value; sidebarAction.value = null }
 function openSidebarAction(kind: 'folders' | 'lists', item: FolderItem | TaskList) {
   sidebarAction.value = sidebarAction.value?.item.id === item.id ? null : { kind, item }
@@ -984,7 +1040,7 @@ onMounted(bootstrap)
         <div v-for="list in lists.filter(l=>!l.folder_id&&!l.is_inbox)" :key="list.id" :data-list-id="list.id" class="list-row" :class="{active:activeView==='tasks'&&activeList===list.id,'list-dragging':listDrag?.id===list.id,'list-reorder-target':listReorderTarget===list.id&&listDrag?.id!==list.id}" :style="{'--list-drag-y':`${listDrag?.id===list.id?listDrag.offsetY:0}px`}" @pointermove="moveListDrag(list,$event)" @pointerup="finishListDrag(list,$event)" @pointercancel="cancelListDrag"><button class="list-drag-handle" aria-label="拖动清单排序或移动到文件夹" title="长按拖动清单" @pointerdown.stop="startListHandlePress(list,$event)" @pointermove.stop="moveListHandle(list,$event)" @pointerup.stop="finishListDrag(list,$event)" @pointercancel.stop="cancelListDrag"><GripVertical/></button><button class="list-row-main" :title="list.name" :aria-label="list.name" @click="selectListUnlessDragged(list)"><i/><span>{{list.name}}</span></button><span class="row-actions"><button aria-label="打开清单操作" :aria-expanded="sidebarAction?.item.id===list.id" @click="openSidebarAction('lists',list)"><Ellipsis/></button></span></div>
         <template v-if="archivedLists.length">
           <div class="section-title"><span>已归档清单</span></div>
-          <div v-for="list in archivedLists" :key="list.id" class="list-row archived-row"><div class="list-row-main archived-row-label" :title="list.name" :aria-label="`已归档清单：${list.name}`"><ArchiveRestore/><span>{{list.name}}</span></div><span class="row-actions archived-actions"><button aria-label="恢复清单" @click="restoreList(list)"><ArchiveRestore/>恢复</button></span></div>
+          <div v-for="list in archivedLists" :key="list.id" class="list-row archived-row"><div class="list-row-main archived-row-label" :title="list.name" :aria-label="`已归档清单：${list.name}`"><ArchiveRestore/><span>{{list.name}}</span></div><span class="row-actions archived-actions"><button aria-label="恢复清单" @click="restoreList(list)"><ArchiveRestore/>恢复</button><button class="danger-text" aria-label="永久删除清单" @click="openPurgeList(list,$event.currentTarget)"><Trash2/>永久删除</button></span></div>
         </template>
       </div>
       <nav class="sidebar-management" aria-label="管理">
@@ -1098,6 +1154,14 @@ onMounted(bootstrap)
     </Transition>
     <Transition name="toast"><div v-if="notice" class="toast" role="status">{{notice}}</div></Transition>
     <div v-if="error" class="error-toast" role="alert">{{error}}<button @click="error=''"><X/></button></div>
+    <div v-if="purgeListTarget" class="modal-mask purge-list-mask" @click.self="closePurgeList">
+      <section ref="purgeListDialog" class="modal-box purge-list-dialog" role="alertdialog" aria-modal="true" aria-labelledby="purge-list-title" aria-describedby="purge-list-description" @keydown="handlePurgeDialogKeydown">
+        <h3 id="purge-list-title">永久删除清单「{{ purgeListTarget.name }}」？</h3>
+        <p id="purge-list-description">将永久删除其中的全部任务、子任务、重复规则、附件及实体文件。此操作无法撤销。</p>
+        <p v-if="purgeListError" role="alert" class="purge-list-error">{{ purgeListError }}</p>
+        <div class="modal-actions"><button ref="purgeCancelButton" class="secondary" :disabled="purgeListSubmitting" @click="closePurgeList">取消</button><button class="danger-button" :disabled="purgeListSubmitting" @click="confirmPurgeList">{{ purgeListSubmitting ? '正在删除…' : '永久删除' }}</button></div>
+      </section>
+    </div>
     <div v-if="modalVisible" class="modal-mask" @click.self="closeModal">
       <div class="modal-box" role="dialog" aria-modal="true">
         <h3>{{ modalTitle }}</h3>

frontend/src/lib/list-purge.test.ts

@@ -0,0 +1,15 @@
+import { describe, expect, it } from 'vitest'
+import { nextDialogFocusIndex } from './list-purge'
+
+describe('archived list purge dialog behavior', () => {
+  it('wraps Tab focus between the cancel and destructive actions', () => {
+    expect(nextDialogFocusIndex(0, 2, true)).toBe(1)
+    expect(nextDialogFocusIndex(1, 2, false)).toBe(0)
+  })
+
+  it('leaves focus alone while moving between interior controls', () => {
+    expect(nextDialogFocusIndex(1, 3, true)).toBeNull()
+    expect(nextDialogFocusIndex(1, 3, false)).toBeNull()
+    expect(nextDialogFocusIndex(0, 0, false)).toBeNull()
+  })
+})

frontend/src/lib/list-purge.ts

@@ -0,0 +1,6 @@
+export function nextDialogFocusIndex(currentIndex: number, controlCount: number, shiftKey: boolean) {
+  if (controlCount < 2) return null
+  if (shiftKey && currentIndex === 0) return controlCount - 1
+  if (!shiftKey && currentIndex === controlCount - 1) return 0
+  return null
+}

frontend/src/style.test.ts

@@ -737,4 +737,32 @@ describe('sidebar layout', () => {
   it('styles archived rows through the new list-row-main structure', () => {
     expect(css).toContain('.archived-row .list-row-main>svg{color:#c9a45c}')
   })
+
+  it('offers permanent deletion only beside archived lists', () => {
+    expect(app).toContain('aria-label="永久删除清单" @click="openPurgeList(list,$event.currentTarget)"')
+    expect(app).toContain("api(`/lists/${purgeListTarget.value.id}/purge`, { method: 'DELETE' })")
+    expect(app).not.toMatch(/list\.is_inbox[^\n]*openPurgeList/)
+  })
+
+  it('uses a guarded custom confirmation that keeps failures visible', () => {
+    expect(app).toContain('将永久删除其中的全部任务、子任务、重复规则、附件及实体文件。此操作无法撤销。')
+    expect(app).toContain('ref="purgeCancelButton"')
+    expect(app).toContain('purgeCancelButton.value?.focus()')
+    expect(app).toContain('@keydown="handlePurgeDialogKeydown"')
+    expect(app).toContain("if (event.key === 'Escape' && !purgeListSubmitting.value) closePurgeList()")
+    expect(app).toContain('if (purgeListSubmitting.value) return')
+    expect(app).toContain('purgeListError.value = reason instanceof Error ? reason.message : \'永久删除失败\'')
+    expect(app).toContain(':disabled="purgeListSubmitting"')
+    expect(app).toContain('role="alert" class="purge-list-error"')
+    expect(css).toContain('.purge-list-dialog button{min-height:44px;')
+  })
+
+  it('removes a purged row and persists inbox navigation after archiving the current list', () => {
+    expect(app).toContain('archivedLists.value = archivedLists.value.filter((list) => list.id !== purgedId)')
+    expect(app).toContain("const wasCurrentList = activeView.value === 'tasks' && activeList.value === item.id")
+    expect(app).toContain('if (wasCurrentList)')
+    expect(app).toContain("await switchView('tasks', inboxId)")
+    expect(app).toContain('selectedTask.value = null')
+    expect(app).toContain("writeStoredNavigation(window.localStorage, NAVIGATION_STORAGE_KEY, view, activeList.value)")
+  })
 })

migrations/versions/0015_purge_operations.py

@@ -0,0 +1,36 @@
+"""track retryable task-list purge cleanup
+
+Revision ID: 0015_purge_operations
+Revises: 0014_preserve_task_due_times
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0015_purge_operations"
+down_revision = "0014_preserve_task_due_times"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "purge_operations",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("list_id", sa.Uuid(), nullable=False),
+        sa.Column("user_id", sa.Uuid(), nullable=False),
+        sa.Column("trash_dir", sa.String(length=1024), nullable=False),
+        sa.Column("status", sa.String(length=32), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(["user_id"], ["users.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("list_id"),
+    )
+    op.create_index("ix_purge_operations_list_id", "purge_operations", ["list_id"])
+    op.create_index("ix_purge_operations_user_id", "purge_operations", ["user_id"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_purge_operations_user_id", table_name="purge_operations")
+    op.drop_index("ix_purge_operations_list_id", table_name="purge_operations")
+    op.drop_table("purge_operations")

tests/test_list_purge.py

@@ -0,0 +1,545 @@
+import asyncio
+import os
+import shutil
+from concurrent.futures import ThreadPoolExecutor
+from datetime import UTC, datetime
+from pathlib import Path
+from uuid import UUID
+
+import pytest
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from backend.config import get_settings
+from backend.db import get_engine
+from backend.models import (
+    Attachment,
+    PurgeOperation,
+    RecurrenceException,
+    RecurrenceTemplate,
+    Task,
+    TaskList,
+    User,
+)
+
+
+def boot(client):
+    response = client.post(
+        "/api/v1/setup/initialize",
+        json={"username": "owner", "password": "correct horse battery staple"},
+    )
+    assert response.status_code == 201
+
+
+def create_list(client, name="待永久删除"):
+    response = client.post("/api/v1/lists", json={"name": name})
+    assert response.status_code == 201
+    return response.json()
+
+
+def attachment_storage_name(attachment_id: str) -> str:
+    async def load():
+        session_factory = async_sessionmaker(get_engine(), expire_on_commit=False)
+        async with session_factory() as db:
+            return (await db.get(Attachment, UUID(attachment_id))).storage_name
+
+    return asyncio.run(load())
+
+
+def test_purge_requires_archive_protects_inbox_and_hides_other_users_lists(client):
+    boot(client)
+    inbox = client.get("/api/v1/lists").json()[0]
+    active = create_list(client, "活动清单")
+
+    assert client.delete(f"/api/v1/lists/{inbox['id']}/purge").status_code == 409
+    active_response = client.delete(f"/api/v1/lists/{active['id']}/purge")
+    assert active_response.status_code == 409
+    assert active_response.json()["detail"] == "请先归档再永久删除"
+
+    async def seed_foreign_archived_list():
+        session_factory = async_sessionmaker(get_engine(), expire_on_commit=False)
+        async with session_factory() as db:
+            foreign = User(username="other", password_hash="unused")
+            db.add(foreign)
+            await db.flush()
+            foreign_list = TaskList(
+                user_id=foreign.id,
+                name="别人的归档清单",
+                deleted_at=datetime.now(UTC),
+            )
+            db.add(foreign_list)
+            await db.commit()
+            return foreign_list.id
+
+    foreign_list_id = asyncio.run(seed_foreign_archived_list())
+    assert client.delete(f"/api/v1/lists/{foreign_list_id}/purge").status_code == 404
+    assert asyncio.run(row_counts(foreign_list_id))["lists"] == 1
+
+
+def test_repeated_purge_is_not_found_and_does_not_touch_other_lists(client):
+    boot(client)
+    target = create_list(client, "目标")
+    survivor = create_list(client, "保留")
+    survivor_task = client.post(
+        "/api/v1/tasks", json={"title": "必须保留", "list_id": survivor["id"]}
+    ).json()
+    assert client.delete(f"/api/v1/lists/{target['id']}").status_code == 204
+    assert client.delete(f"/api/v1/lists/{target['id']}/purge").status_code == 204
+
+    assert client.delete(f"/api/v1/lists/{target['id']}/purge").status_code == 404
+    assert client.get(f"/api/v1/tasks/{survivor_task['id']}").status_code == 200
+
+
+def test_purge_uses_unique_operation_directory_and_preserves_legacy_collision(
+    client, tmp_path: Path
+):
+    boot(client)
+    attachment_root = tmp_path / "attachments"
+    get_settings().attachment_dir = str(attachment_root)
+    task_list = create_list(client)
+    task = client.post(
+        "/api/v1/tasks", json={"title": "带附件", "list_id": task_list["id"]}
+    ).json()
+    upload = client.post(
+        f"/api/v1/tasks/{task['id']}/attachments",
+        files={"file": ("proof.txt", b"proof", "text/plain")},
+    )
+    assert upload.status_code == 201
+    legacy_collision = attachment_root / ".purge-trash" / task_list["id"]
+    legacy_collision.mkdir(parents=True)
+    sentinel = legacy_collision / "owned-by-another-request"
+    sentinel.write_text("keep", encoding="utf-8")
+    assert client.delete(f"/api/v1/lists/{task_list['id']}").status_code == 204
+
+    response = client.delete(f"/api/v1/lists/{task_list['id']}/purge")
+
+    assert response.status_code == 204
+    assert sentinel.read_text(encoding="utf-8") == "keep"
+    assert asyncio.run(row_counts(task_list["id"]))["lists"] == 0
+
+
+def test_operation_directory_collision_does_not_delete_unowned_directory(
+    client, tmp_path: Path, monkeypatch
+):
+    import backend.main as main_module
+
+    boot(client)
+    attachment_root = tmp_path / "attachments"
+    get_settings().attachment_dir = str(attachment_root)
+    task_list = create_list(client)
+    task = client.post(
+        "/api/v1/tasks", json={"title": "带附件", "list_id": task_list["id"]}
+    ).json()
+    client.post(
+        f"/api/v1/tasks/{task['id']}/attachments",
+        files={"file": ("proof.txt", b"proof", "text/plain")},
+    )
+    operation_id = UUID("aaaaaaaa-aaaa-4aaa-aaaa-aaaaaaaaaaaa")
+    collision = attachment_root / ".purge-trash" / str(operation_id)
+    collision.mkdir(parents=True)
+    sentinel = collision / "owned-by-another-request"
+    sentinel.write_text("keep", encoding="utf-8")
+    operation_ids = iter(
+        [operation_id, UUID("bbbbbbbb-bbbb-4bbb-bbbb-bbbbbbbbbbbb")]
+    )
+    monkeypatch.setattr(main_module, "uuid4", lambda: next(operation_ids))
+    assert client.delete(f"/api/v1/lists/{task_list['id']}").status_code == 204
+
+    response = client.delete(f"/api/v1/lists/{task_list['id']}/purge")
+
+    assert response.status_code == 204
+    assert sentinel.read_text(encoding="utf-8") == "keep"
+    assert asyncio.run(row_counts(task_list["id"]))["lists"] == 0
+
+
+def test_concurrent_quarantine_allocation_uses_distinct_owned_directories(tmp_path: Path):
+    from backend.main import _create_unique_quarantine
+
+    attachment_root = tmp_path / "attachments"
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        allocations = list(pool.map(lambda _: _create_unique_quarantine(attachment_root), range(20)))
+
+    operation_ids = [operation_id for operation_id, _ in allocations]
+    trash_dirs = [trash_dir for _, trash_dir in allocations]
+    assert len(set(operation_ids)) == 20
+    assert len(set(trash_dirs)) == 20
+    assert all(trash_dir.is_dir() for trash_dir in trash_dirs)
+
+
+def test_attachment_pre_move_failure_keeps_database_and_file(
+    client, tmp_path: Path, monkeypatch
+):
+    boot(client)
+    get_settings().attachment_dir = str(tmp_path / "attachments")
+    task_list = create_list(client)
+    task = client.post(
+        "/api/v1/tasks", json={"title": "带附件", "list_id": task_list["id"]}
+    ).json()
+    upload = client.post(
+        f"/api/v1/tasks/{task['id']}/attachments",
+        files={"file": ("proof.txt", b"proof", "text/plain")},
+    )
+    attachment_path = Path(get_settings().attachment_dir) / attachment_storage_name(upload.json()["id"])
+    assert client.delete(f"/api/v1/lists/{task_list['id']}").status_code == 204
+
+    def fail_replace(source, destination):
+        raise OSError("disk unavailable")
+
+    monkeypatch.setattr(os, "replace", fail_replace)
+    response = client.delete(f"/api/v1/lists/{task_list['id']}/purge")
+
+    assert response.status_code == 500
+    assert response.json()["detail"] == "附件隔离失败，清单未删除"
+    assert asyncio.run(row_counts(task_list["id"]))["lists"] == 1
+    assert asyncio.run(row_counts(task_list["id"]))["attachments"] == 1
+    assert attachment_path.read_bytes() == b"proof"
+
+
+def test_pre_move_restore_failure_preserves_quarantined_copy(
+    client, tmp_path: Path, monkeypatch
+):
+    boot(client)
+    attachment_root = tmp_path / "attachments"
+    get_settings().attachment_dir = str(attachment_root)
+    task_list = create_list(client)
+    task = client.post(
+        "/api/v1/tasks", json={"title": "两个附件", "list_id": task_list["id"]}
+    ).json()
+    first = client.post(
+        f"/api/v1/tasks/{task['id']}/attachments",
+        files={"file": ("one.txt", b"one", "text/plain")},
+    ).json()
+    second = client.post(
+        f"/api/v1/tasks/{task['id']}/attachments",
+        files={"file": ("two.txt", b"two", "text/plain")},
+    ).json()
+    first_name = attachment_storage_name(first["id"])
+    second_name = attachment_storage_name(second["id"])
+    assert client.delete(f"/api/v1/lists/{task_list['id']}").status_code == 204
+    original_replace = os.replace
+    calls = 0
+
+    def fail_second_move_and_restore(source, destination):
+        nonlocal calls
+        calls += 1
+        if calls >= 2:
+            raise OSError("disk unavailable")
+        return original_replace(source, destination)
+
+    monkeypatch.setattr(os, "replace", fail_second_move_and_restore)
+    response = client.delete(f"/api/v1/lists/{task_list['id']}/purge")
+
+    assert response.status_code == 500
+    assert "部分附件恢复失败" in response.json()["detail"]
+    trash_dirs = list((attachment_root / ".purge-trash").iterdir())
+    assert len(trash_dirs) == 1
+    assert (trash_dirs[0] / first_name).read_bytes() == b"one"
+    assert (attachment_root / second_name).read_bytes() == b"two"
+    assert asyncio.run(row_counts(task_list["id"]))["lists"] == 1
+    assert asyncio.run(row_counts(task_list["id"]))["attachments"] == 2
+
+
+def test_database_commit_failure_restores_quarantined_file_and_rows(
+    client, tmp_path: Path, monkeypatch
+):
+    boot(client)
+    get_settings().attachment_dir = str(tmp_path / "attachments")
+    task_list = create_list(client)
+    task = client.post(
+        "/api/v1/tasks", json={"title": "带附件", "list_id": task_list["id"]}
+    ).json()
+    upload = client.post(
+        f"/api/v1/tasks/{task['id']}/attachments",
+        files={"file": ("proof.txt", b"proof", "text/plain")},
+    )
+    attachment_path = Path(get_settings().attachment_dir) / attachment_storage_name(upload.json()["id"])
+    assert client.delete(f"/api/v1/lists/{task_list['id']}").status_code == 204
+    original_commit = AsyncSession.commit
+
+    async def fail_commit(db):
+        raise OSError("database unavailable")
+
+    monkeypatch.setattr(AsyncSession, "commit", fail_commit)
+    response = client.delete(f"/api/v1/lists/{task_list['id']}/purge")
+    monkeypatch.setattr(AsyncSession, "commit", original_commit)
+
+    assert response.status_code == 500
+    assert response.json()["detail"] == "数据库提交失败，清单未删除"
+    assert asyncio.run(row_counts(task_list["id"]))["lists"] == 1
+    assert asyncio.run(row_counts(task_list["id"]))["attachments"] == 1
+    assert attachment_path.read_bytes() == b"proof"
+    assert not (Path(get_settings().attachment_dir) / ".purge-trash").exists()
+
+
+def test_cleanup_failure_can_be_retried_by_original_list_id(
+    client, tmp_path: Path, monkeypatch
+):
+    boot(client)
+    attachment_root = tmp_path / "attachments"
+    get_settings().attachment_dir = str(attachment_root)
+    task_list = create_list(client)
+    task = client.post(
+        "/api/v1/tasks", json={"title": "带附件", "list_id": task_list["id"]}
+    ).json()
+    upload = client.post(
+        f"/api/v1/tasks/{task['id']}/attachments",
+        files={"file": ("proof.txt", b"proof", "text/plain")},
+    )
+    storage_name = attachment_storage_name(upload.json()["id"])
+    assert client.delete(f"/api/v1/lists/{task_list['id']}").status_code == 204
+    original_rmtree = shutil.rmtree
+    cleanup_attempts = 0
+
+    def fail_cleanup_once(path, *args, **kwargs):
+        nonlocal cleanup_attempts
+        if Path(path).parent.name == ".purge-trash" and cleanup_attempts == 0:
+            cleanup_attempts += 1
+            raise OSError("disk unavailable")
+        return original_rmtree(path, *args, **kwargs)
+
+    monkeypatch.setattr(shutil, "rmtree", fail_cleanup_once)
+    first = client.delete(f"/api/v1/lists/{task_list['id']}/purge")
+    assert first.status_code == 500
+    operation_id = first.json()["detail"].split("操作ID: ", 1)[1].split(",", 1)[0]
+    trash_dir = attachment_root / ".purge-trash" / operation_id
+    assert (trash_dir / storage_name).read_bytes() == b"proof"
+
+    second = client.delete(f"/api/v1/lists/{task_list['id']}/purge")
+
+    assert second.status_code == 204
+    assert not trash_dir.exists()
+    assert asyncio.run(purge_operation_count(task_list["id"])) == 0
+
+
+async def purge_operation_count(list_id):
+    session_factory = async_sessionmaker(get_engine(), expire_on_commit=False)
+    async with session_factory() as db:
+        return await db.scalar(
+            select(func.count()).select_from(PurgeOperation).where(
+                PurgeOperation.list_id == UUID(str(list_id))
+            )
+        )
+
+
+def test_final_quarantine_cleanup_failure_is_reported_and_keeps_residue_isolated(
+    client, tmp_path: Path, monkeypatch, caplog
+):
+    boot(client)
+    attachment_root = tmp_path / "attachments"
+    get_settings().attachment_dir = str(attachment_root)
+    task_list = create_list(client)
+    task = client.post(
+        "/api/v1/tasks", json={"title": "带附件", "list_id": task_list["id"]}
+    ).json()
+    upload = client.post(
+        f"/api/v1/tasks/{task['id']}/attachments",
+        files={"file": ("proof.txt", b"proof", "text/plain")},
+    )
+    storage_name = attachment_storage_name(upload.json()["id"])
+    attachment_path = attachment_root / storage_name
+    assert client.delete(f"/api/v1/lists/{task_list['id']}").status_code == 204
+    original_rmtree = shutil.rmtree
+
+    def fail_trash_cleanup(path, *args, **kwargs):
+        if Path(path).parent.name == ".purge-trash":
+            raise OSError("disk unavailable")
+        return original_rmtree(path, *args, **kwargs)
+
+    monkeypatch.setattr(shutil, "rmtree", fail_trash_cleanup)
+    with caplog.at_level("ERROR", logger="backend.main"):
+        response = client.delete(f"/api/v1/lists/{task_list['id']}/purge")
+
+    trash_dirs = [path for path in (attachment_root / ".purge-trash").iterdir()]
+    assert len(trash_dirs) == 1
+    trash_dir = trash_dirs[0]
+    assert response.status_code == 500
+    assert "清单数据已删除，但附件清理未完成" in response.json()["detail"]
+    assert f"操作ID: {trash_dir.name}" in response.json()["detail"]
+    assert asyncio.run(row_counts(task_list["id"]))["lists"] == 0
+    assert not attachment_path.exists()
+    assert (trash_dir / storage_name).read_bytes() == b"proof"
+    assert str(trash_dir) in caplog.text
+
+
+def test_malicious_attachment_path_rejects_purge_without_touching_file_or_database(
+    client, tmp_path: Path
+):
+    boot(client)
+    attachment_root = tmp_path / "attachments"
+    get_settings().attachment_dir = str(attachment_root)
+    task_list = create_list(client)
+    task = client.post(
+        "/api/v1/tasks", json={"title": "恶意附件", "list_id": task_list["id"]}
+    ).json()
+    outside_path = tmp_path / "outside.txt"
+    outside_path.write_text("keep", encoding="utf-8")
+
+    async def seed_malicious_attachment():
+        session_factory = async_sessionmaker(get_engine(), expire_on_commit=False)
+        async with session_factory() as db:
+            owner_id = await db.scalar(select(TaskList.user_id).where(TaskList.id == UUID(task_list["id"])))
+            db.add(
+                Attachment(
+                    user_id=owner_id,
+                    task_id=UUID(task["id"]),
+                    filename="outside.txt",
+                    storage_name="../outside.txt",
+                    mime_type="text/plain",
+                    size=4,
+                )
+            )
+            await db.commit()
+
+    asyncio.run(seed_malicious_attachment())
+    assert client.delete(f"/api/v1/lists/{task_list['id']}").status_code == 204
+
+    response = client.delete(f"/api/v1/lists/{task_list['id']}/purge")
+
+    assert response.status_code == 409
+    assert response.json()["detail"] == "附件存储路径无效，无法永久删除清单"
+    assert outside_path.read_text(encoding="utf-8") == "keep"
+    assert asyncio.run(row_counts(task_list["id"]))["lists"] == 1
+    assert asyncio.run(row_counts(task_list["id"]))["attachments"] == 1
+
+
+@pytest.mark.parametrize(
+    "storage_name",
+    [
+        "/tmp/dodo-outside.txt",
+        "nested/../proof.txt",
+        "nested/./proof.txt",
+        "nested//proof.txt",
+        "C:\\temp\\proof.txt",
+    ],
+)
+def test_absolute_navigation_or_empty_attachment_path_segment_is_rejected(
+    client, tmp_path: Path, storage_name: str
+):
+    boot(client)
+    get_settings().attachment_dir = str(tmp_path / "attachments")
+    task_list = create_list(client)
+    task = client.post(
+        "/api/v1/tasks", json={"title": "恶意附件", "list_id": task_list["id"]}
+    ).json()
+
+    async def seed_malicious_attachment():
+        session_factory = async_sessionmaker(get_engine(), expire_on_commit=False)
+        async with session_factory() as db:
+            owner_id = await db.scalar(select(TaskList.user_id).where(TaskList.id == UUID(task_list["id"])))
+            db.add(
+                Attachment(
+                    user_id=owner_id,
+                    task_id=UUID(task["id"]),
+                    filename="proof.txt",
+                    storage_name=storage_name,
+                    mime_type="text/plain",
+                    size=4,
+                )
+            )
+            await db.commit()
+
+    asyncio.run(seed_malicious_attachment())
+    assert client.delete(f"/api/v1/lists/{task_list['id']}").status_code == 204
+    assert client.delete(f"/api/v1/lists/{task_list['id']}/purge").status_code == 409
+    assert asyncio.run(row_counts(task_list["id"]))["lists"] == 1
+
+
+async def row_counts(list_id):
+    list_id = UUID(str(list_id))
+    session_factory = async_sessionmaker(get_engine(), expire_on_commit=False)
+    async with session_factory() as db:
+        task_ids = list((await db.scalars(select(Task.id).where(Task.list_id == list_id))).all())
+        template_ids = list(
+            (await db.scalars(select(RecurrenceTemplate.id).where(RecurrenceTemplate.task_id.in_(task_ids)))).all()
+        ) if task_ids else []
+        return {
+            "lists": await db.scalar(select(func.count()).select_from(TaskList).where(TaskList.id == list_id)),
+            "tasks": len(task_ids),
+            "templates": len(template_ids),
+            "exceptions": await db.scalar(
+                select(func.count()).select_from(RecurrenceException).where(
+                    RecurrenceException.template_id.in_(template_ids)
+                )
+            ) if template_ids else 0,
+            "attachments": await db.scalar(
+                select(func.count()).select_from(Attachment).where(Attachment.task_id.in_(task_ids))
+            ) if task_ids else 0,
+        }
+
+
+def test_purge_archived_list_removes_all_task_data_and_attachment_files(client, tmp_path: Path):
+    boot(client)
+    get_settings().attachment_dir = str(tmp_path / "attachments")
+    task_list = create_list(client)
+    parent = client.post(
+        "/api/v1/tasks",
+        json={
+            "title": "重复父任务",
+            "list_id": task_list["id"],
+            "due_at": "2030-01-01T09:00:00Z",
+            "rrule": "FREQ=DAILY",
+        },
+    ).json()
+    child = client.post(
+        "/api/v1/tasks",
+        json={"title": "子任务", "list_id": task_list["id"], "parent_id": parent["id"]},
+    ).json()
+    completed = client.post(
+        "/api/v1/tasks", json={"title": "已完成", "list_id": task_list["id"]}
+    ).json()
+    completed = client.patch(
+        f"/api/v1/tasks/{completed['id']}", json={"completed": True, "version": completed["version"]}
+    ).json()
+    soft_deleted = client.post(
+        "/api/v1/tasks", json={"title": "软删除", "list_id": task_list["id"]}
+    ).json()
+    assert client.delete(f"/api/v1/tasks/{soft_deleted['id']}").status_code == 204
+
+    upload = client.post(
+        f"/api/v1/tasks/{child['id']}/attachments",
+        files={"file": ("proof.txt", b"proof", "text/plain")},
+    )
+    assert upload.status_code == 201
+    attachment_id = upload.json()["id"]
+
+    async def seed_exception_and_get_storage_name():
+        session_factory = async_sessionmaker(get_engine(), expire_on_commit=False)
+        async with session_factory() as db:
+            template = await db.scalar(
+                select(RecurrenceTemplate).where(RecurrenceTemplate.task_id == UUID(parent["id"]))
+            )
+            db.add(
+                RecurrenceException(
+                    template_id=template.id,
+                    occurrence_at=datetime(2030, 1, 2, 9, tzinfo=UTC),
+                    completed=True,
+                )
+            )
+            attachment = await db.get(Attachment, UUID(attachment_id))
+            storage_name = attachment.storage_name
+            await db.commit()
+            return storage_name
+
+    storage_name = asyncio.run(seed_exception_and_get_storage_name())
+    attachment_path = Path(get_settings().attachment_dir) / storage_name
+    assert attachment_path.is_file()
+    assert asyncio.run(row_counts(task_list["id"])) == {
+        "lists": 1,
+        "tasks": 4,
+        "templates": 1,
+        "exceptions": 1,
+        "attachments": 1,
+    }
+
+    assert client.delete(f"/api/v1/lists/{task_list['id']}").status_code == 204
+    response = client.delete(f"/api/v1/lists/{task_list['id']}/purge")
+
+    assert response.status_code == 204
+    assert asyncio.run(row_counts(task_list["id"])) == {
+        "lists": 0,
+        "tasks": 0,
+        "templates": 0,
+        "exceptions": 0,
+        "attachments": 0,
+    }
+    assert not attachment_path.exists()