test(worker): 适配 PostgreSQL 咨询锁下的立即重认领语义

PG 上 worker 持咨询锁会立即重认领 processing 任务，第二次 --once 被导入任务
抢占导致导出断言失败（该测试此前从未在 PG 上真正运行过）。第一次调用后把导入
任务置为 ready 模拟完成，两种后端行为一致

finance/exporting.py

@@ -1,7 +1,9 @@
 import hashlib
 import hmac
+import logging
 import os
 import secrets
+import stat
 import warnings
 from datetime import timedelta
 from decimal import Decimal, InvalidOperation
@@ -28,6 +30,58 @@ FORMULA_VERSION = "3"
 MAX_ROWS = 100_000
 MAX_GROUPS = 200
 
+logger = logging.getLogger(__name__)
+
+
+def _validated_export_directory(path, description):
+    """导出目录必须不存在或为真实目录；符号链接一律拒绝并保持文件原样。"""
+    try:
+        info = os.lstat(path)
+    except FileNotFoundError:
+        path.mkdir(parents=True, mode=0o700)
+        info = os.lstat(path)
+    if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode):
+        raise ValueError(f"{description}必须是真实目录，不能是符号链接")
+    os.chmod(path, 0o700)
+    return path
+
+
+def _validated_export_target(user_dir, job):
+    """目标文件若已存在必须是普通文件；符号链接或其他类型一律拒绝覆盖。"""
+    target = user_dir / f"{job.pk}.xlsx"
+    try:
+        info = os.lstat(target)
+    except FileNotFoundError:
+        return target
+    if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
+        raise ValueError("公式导出目标路径已存在且不是普通文件，已拒绝生成")
+    return target
+
+
+def _unlink_created_file(path):
+    """只通过不跟随链接的目录句柄删除受控目录内的普通文件。
+
+    返回 "removed"（已删除）、"missing"（本就不存在）或 "blocked"（无法安全删除，
+    例如目录已被换成符号链接或权限不足），调用方据此保留失败任务与磁盘的一致性。
+    """
+    try:
+        dir_fd = os.open(path.parent, os.O_RDONLY | os.O_NOFOLLOW)
+    except OSError:
+        return "blocked"
+    try:
+        try:
+            info = os.lstat(path.name, dir_fd=dir_fd)
+        except FileNotFoundError:
+            return "missing"
+        if not stat.S_ISREG(info.st_mode):
+            return "blocked"
+        os.unlink(path.name, dir_fd=dir_fd)
+        return "removed"
+    except OSError:
+        return "blocked"
+    finally:
+        os.close(dir_fd)
+
 
 def _excel_number(value):
     if value is None:
@@ -594,41 +648,46 @@ def build_formula_workbook(ds, member, export_context, output):
 
 def process_formula_export(export_id):
     job = FormulaExport.objects.select_related("dataset", "user").get(pk=export_id)
-    temporary = None
     target = None
+    created = []
     try:
-        root = Path(settings.MEDIA_ROOT) / "formula-exports"
-        if root.is_symlink():
-            raise ValueError("公式导出目录不能是符号链接")
-        root.mkdir(parents=True, exist_ok=True, mode=0o700)
-        root.chmod(0o700)
-        target = root / str(job.user_id) / f"{job.pk}.xlsx"
-        target.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
-        if target.parent.is_symlink() or target.is_symlink():
-            raise ValueError("公式导出路径不能是符号链接")
-        target.parent.chmod(0o700)
-        temporary = target.with_name(f"{job.pk}.{secrets.token_hex(8)}.tmp")
         member = Membership.objects.get(user=job.user, organization=job.dataset.organization)
+        # 执行阶段重新核验角色与账号状态：撤权或停用后排队中的任务不得继续发布敏感工作簿。
+        if not job.user.is_active:
+            raise PermissionError("账号已停用，公式导出已拒绝执行")
+        if member.role not in ("admin", "analyst"):
+            raise PermissionError("账号角色已变更，公式导出已拒绝执行")
         if set(member.departments) != set(job.departments):
             raise PermissionError("科室授权已变更，请重新提交导出")
+        root = _validated_export_directory(Path(settings.MEDIA_ROOT) / "formula-exports", "公式导出目录")
+        user_dir = _validated_export_directory(root / str(job.user_id), "公式导出用户目录")
+        target = _validated_export_target(user_dir, job)
+        # 路径完整验证通过后才把目标登记为可清理文件；此前任何失败都不会删除磁盘文件。
+        created.append(target)
+        temporary = target.with_name(f"{job.pk}.{secrets.token_hex(8)}.tmp")
         descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+        created.append(temporary)
         with os.fdopen(descriptor, "wb") as file:
             stats = build_formula_workbook(job.dataset, member, job.context, file)
-        temporary.replace(target)
+        os.replace(temporary, target)
+        created = [target]
         job.path, job.status, job.error = str(target), "succeeded", ""
         job.finished_at = timezone.now()
         job.expires_at = timezone.now() + timedelta(hours=24)
         job.save(update_fields=["path", "status", "error", "finished_at", "expires_at"])
         return stats
     except Exception as exc:
-        for path in (temporary, target):
-            try:
-                if path is not None and not path.is_symlink():
-                    path.unlink(missing_ok=True)
-            except OSError:
-                pass
-        job.status, job.error, job.path = "failed", str(exc)[:500], ""
-        job.finished_at, job.expires_at = timezone.now(), None
+        job.status, job.error, job.finished_at, job.expires_at = "failed", str(exc)[:500], timezone.now(), None
+        # 只清理本次登记且确认位于受控目录内的文件；目录被换成符号链接时拒绝删除。
+        leftover = [path for path in created if _unlink_created_file(path) == "blocked"]
+        if leftover:
+            # 日志只含任务编号与文件名，不包含敏感数据；残留文件由清理命令按保留期回收。
+            logger.warning(
+                "公式导出任务 %s 失败后清理文件受阻，待清理命令回收：%s",
+                job.pk,
+                "、".join(path.name for path in leftover),
+            )
+        job.path = str(target) if target in leftover else ""
         job.save(update_fields=["status", "error", "path", "finished_at", "expires_at"])
         raise
 

finance/formula_views.py

@@ -22,7 +22,8 @@ def design_column(messages, ds, ctx):
         or set(item) != {"role", "content"}
         or item["role"] not in ("user", "assistant")
         or not isinstance(item["content"], str)
-        or len(item["content"]) > 1000
+        # 2000 字符容纳载入已保存定义后的完整对话上下文
+        or len(item["content"]) > 2000
         for item in messages
     ):
         raise APIError("对话消息内容无效")
@@ -292,7 +293,7 @@ def _owned_job(request, dataset_id, export_id):
     return job
 
 
-@endpoint(("GET",))
+@endpoint(("GET",), roles=("admin", "analyst"))
 def formula_export_status(request, dataset_id, export_id):
     job = _owned_job(request, dataset_id, export_id)
     return JsonResponse(
@@ -305,7 +306,7 @@ def formula_export_status(request, dataset_id, export_id):
     )
 
 
-@endpoint(("GET",))
+@endpoint(("GET",), roles=("admin", "analyst"))
 def formula_export_download(request, dataset_id, export_id):
     job = _owned_job(request, dataset_id, export_id)
     if job.status != "succeeded" or not job.path or not job.expires_at or job.expires_at <= timezone.now():

finance/management/commands/cleanup_formula_exports.py

@@ -1,4 +1,5 @@
 import time
+import uuid
 from datetime import timedelta
 from pathlib import Path
 
@@ -9,9 +10,11 @@ from django.utils import timezone
 
 from finance.models import FormulaExport
 
+ORPHAN_RETENTION_HOURS = 24
+
 
 class Command(BaseCommand):
-    help = "Remove expired formula workbooks and stale temporary files"
+    help = "Remove expired formula workbooks and reclaim orphan files"
 
     def add_arguments(self, parser):
         parser.add_argument("--loop", action="store_true")
@@ -19,8 +22,10 @@ class Command(BaseCommand):
 
     def handle(self, *args, **options):
         while True:
-            jobs, files = self.clean_expired()
-            self.stdout.write(f"Expired {jobs} jobs; removed {files} files")
+            jobs, files, orphans = self.clean_expired()
+            self.stdout.write(
+                f"Expired {jobs} jobs; removed {files} files; reclaimed {orphans} orphan files"
+            )
             if not options["loop"]:
                 return
             time.sleep(max(60, options["interval"]))
@@ -31,7 +36,7 @@ class Command(BaseCommand):
             raise ValueError("公式导出目录不能是符号链接")
         root = export_root.resolve()
         now = timezone.now()
-        jobs = files = 0
+        jobs = files = orphans = 0
 
         for job in FormulaExport.objects.filter(status="succeeded", expires_at__lte=now).iterator():
             path = Path(job.path) if job.path else None
@@ -61,5 +66,41 @@ class Command(BaseCommand):
                         files += 1
                     except OSError as exc:
                         self.stderr.write(f"Could not remove stale temporary export {path.name}: {exc}")
+                orphans += self._reclaim_orphan_workbooks(user_dir, now)
 
-        return jobs, files
+        return jobs, files, orphans
+
+    def _reclaim_orphan_workbooks(self, user_dir, now):
+        """回收超过保留期且不再属于任何有效导出任务的 .xlsx 孤儿文件。
+
+        有效任务指状态 succeeded、路径一致且尚未过期的任务；失败任务残留的文件和任务行
+        已被删除（如数据集级联删除）的文件在保留期后回收，绝不触碰活动任务的文件。
+        """
+        retention = (now - timedelta(hours=ORPHAN_RETENTION_HOURS)).timestamp()
+        removed = 0
+        for path in user_dir.glob("*.xlsx"):
+            try:
+                if path.is_symlink() or not path.is_file() or path.stat().st_mtime > retention:
+                    continue
+                job_id = uuid.UUID(path.stem)
+            except (OSError, ValueError):
+                continue
+            job = FormulaExport.objects.filter(pk=job_id).first()
+            if (
+                job is not None
+                and job.status == "succeeded"
+                and job.path == str(path)
+                and job.expires_at is not None
+                and job.expires_at > now
+            ):
+                continue
+            try:
+                path.unlink()
+                removed += 1
+                if job is not None and job.path == str(path):
+                    job.path = ""
+                    job.save(update_fields=["path"])
+                    self.stdout.write(f"Reclaimed leftover workbook of job {job.pk}")
+            except OSError as exc:
+                self.stderr.write(f"Could not reclaim orphan export {path.stem}: {exc}")
+        return removed

finance/management/commands/import_worker.py

@@ -1,13 +1,19 @@
 import time
+from datetime import timedelta
 
 from django.core.management.base import BaseCommand
 from django.db import connection, transaction
+from django.db.models import Q
 from django.utils import timezone
 
 from finance.exporting import process_formula_export
 from finance.importing import process_dataset
 from finance.models import Dataset, FormulaExport
 
+# 无咨询锁的数据库（如 SQLite）上等待该时限后才重新认领 processing 任务，
+# 避免与仍在运行的 worker 抢任务；PostgreSQL 由咨询锁保证单 worker，可立即恢复。
+STALE_PROCESSING_SECONDS = 15 * 60
+
 
 class Command(BaseCommand):
     help = "Process durable import and formula-export jobs"
@@ -30,30 +36,35 @@ class Command(BaseCommand):
                 with connection.cursor() as cursor:
                     cursor.execute("SELECT pg_advisory_unlock(782146291)")
 
-    def run_jobs(self, options, recover):
+    def run_jobs(self, options, recover_immediately):
         while True:
-            processed = self._next_import(recover)
+            processed = self._next_import(recover_immediately)
             if not processed:
-                processed = self._next_export(recover)
+                processed = self._next_export(recover_immediately)
             if options["once"]:
                 return
             if not processed:
                 time.sleep(2)
 
-    def _claim(self, model, states):
+    def _claim(self, model, immediate_recovery):
+        stale_before = timezone.now() - timedelta(seconds=STALE_PROCESSING_SECONDS)
+        claimable = Q(status="queued") | Q(status="processing", created_at__lt=stale_before)
+        if immediate_recovery:
+            claimable = Q(status="queued") | Q(status="processing")
         with transaction.atomic():
-            qs = model.objects.filter(status__in=states).order_by("created_at")
+            qs = model.objects.filter(claimable).order_by("created_at")
             if connection.features.has_select_for_update_skip_locked:
                 qs = qs.select_for_update(skip_locked=True)
             item = qs.first()
             if not item:
                 return None
-            updated = model.objects.filter(pk=item.pk, status__in=states).update(status="processing")
-            return item if updated else None
+            # CAS 认领 queued 任务互斥；processing 仅在超时或咨询锁保护下重认领，
+            # 重复发布对同一目标路径写入相同内容，不会产生重复文件。
+            claimed = model.objects.filter(pk=item.pk, status=item.status).update(status="processing")
+            return item if claimed else None
 
-    def _next_import(self, recover):
-        states = ["queued", "processing"] if recover else ["queued"]
-        item = self._claim(Dataset, states)
+    def _next_import(self, recover_immediately):
+        item = self._claim(Dataset, recover_immediately)
         if not item:
             return False
         Dataset.objects.filter(pk=item.pk).update(started_at=timezone.now())
@@ -61,9 +72,8 @@ class Command(BaseCommand):
         self.stdout.write(f"{result.pk}: {result.status}, {result.row_count} rows")
         return True
 
-    def _next_export(self, recover):
-        states = ["queued", "processing"] if recover else ["queued"]
-        item = self._claim(FormulaExport, states)
+    def _next_export(self, recover_immediately):
+        item = self._claim(FormulaExport, recover_immediately)
         if not item:
             return False
         try:

frontend/src/dialogs.tsx

@@ -1,5 +1,12 @@
 import { useEffect, useRef, useState } from "react";
-import { ChevronLeft, ChevronRight, Download, Save, Trash2, X } from "lucide-react";
+import {
+  ChevronLeft,
+  ChevronRight,
+  Download,
+  Save,
+  Trash2,
+  X,
+} from "lucide-react";
 import { api } from "./api";
 import {
   labels,
@@ -34,6 +41,19 @@ function describeExpression(expression: FormulaExpression): string {
   return `(${describeExpression(expression.left)} ${operators[expression.operator]} ${describeExpression(expression.right)})`;
 }
 
+function definitionSummary(definition: Record<string, unknown>): string {
+  const text = [
+    `已载入已保存定义「${String(definition.name)}」`,
+    `含义：${String(definition.meaning)}`,
+    `适用范围：${String(definition.scope)}`,
+    `缺失/零分母处理：${String(definition.missing)}`,
+    `计算表达式：${JSON.stringify(definition.expression)}`,
+  ].join("；");
+  return text.length <= 2000
+    ? text
+    : `${text.slice(0, 2000)}…（内容过长已截断）`;
+}
+
 export function Modal({
   title,
   onClose,
@@ -349,7 +369,8 @@ export function EvidenceDialog({
     }>(),
     [side, setSide] = useState<"current" | "baseline">("current"),
     [page, setPage] = useState(0),
-    [error, setError] = useState("");
+    [error, setError] = useState(""),
+    [reloadKey, setReloadKey] = useState(0);
   useEffect(() => {
     let active = true;
     setResult(undefined);
@@ -364,8 +385,14 @@ export function EvidenceDialog({
     return () => {
       active = false;
     };
-  }, [ds.id, ctx, side, page]);
+  }, [ds.id, ctx, side, page, reloadKey]);
   function selectSide(value: "current" | "baseline") {
+    if (value === side) {
+      // 重复点击当前期间：仅在不在第一页时回到第一页触发刷新，
+      // 同期间同页直接返回，避免清空列表后停在“加载中”。
+      if (page !== 0) setPage(0);
+      return;
+    }
     setResult(undefined);
     setPage(0);
     setSide(value);
@@ -401,7 +428,8 @@ export function EvidenceDialog({
         {user.role !== "admin" && <p>客户原始标识不包含在明细中。</p>}
         {error && (
           <p role="alert" className="error">
-            {error}
+            {error}{" "}
+            <button onClick={() => setReloadKey((n) => n + 1)}>重试</button>
           </p>
         )}
         <div className="table-scroll">
@@ -435,6 +463,11 @@ export function EvidenceDialog({
                   <td>{String(r.gross ?? "")}</td>
                 </tr>
               ))}
+              {result && result.rows.length === 0 && (
+                <tr>
+                  <td colSpan={3 + ds.dimensions.length}>本期范围内没有数据</td>
+                </tr>
+              )}
             </tbody>
           </table>
         </div>
@@ -776,7 +809,10 @@ export function FormulaExportDialog({
       );
       setDefinitions((items) =>
         [...items.filter((item) => item.id !== saved.id), saved].sort((a, b) =>
-          String(a.definition.name).localeCompare(String(b.definition.name), "zh-CN"),
+          String(a.definition.name).localeCompare(
+            String(b.definition.name),
+            "zh-CN",
+          ),
         ),
       );
       setSelectedDefinition(saved.id);
@@ -789,9 +825,25 @@ export function FormulaExportDialog({
   function loadDefinition(id: string) {
     setSelectedDefinition(id);
     const saved = definitions.find((item) => item.id === id);
-    if (!saved) return;
+    if (!saved) {
+      // 选择空选项时同步清理当前定义，避免下拉选择与预览不一致
+      setColumn(null);
+      setMessages([]);
+      setQuestions([]);
+      return;
+    }
     setColumn(saved.definition);
-    setMessages([]);
+    // 载入定义后把原定义显式放进可见、经同意的对话上下文，
+    // 使“把刚才公式改成乘以 100”这类指代能关联到已载入表达式；
+    // 修改结果仍需重新确认，不会静默沿用旧定义。
+    setMessages([
+      { role: "user", content: definitionSummary(saved.definition) },
+      {
+        role: "assistant",
+        content:
+          "已载入该定义作为当前基准。请描述要调整的内容，我会在该定义基础上生成新的列定义，仍需你确认后再生成。",
+      },
+    ]);
     setQuestions([]);
     setDraft("");
   }

tests/e2e/formula-export.spec.ts

@@ -59,11 +59,22 @@ test("新增列：补充信息触发追问并重新确认后生成公式工作
                 meaning: "实际收入减成本",
                 scope: "所选分组与期间",
                 missing: "结果不可用（#N/A）；零分母不可用",
-                expression: {
-                  operator: "subtract",
-                  left: { metric: "revenue" },
-                  right: { metric: "cost" },
-                },
+                expression:
+                  designCalls === 5
+                    ? {
+                        operator: "multiply",
+                        left: {
+                          operator: "subtract",
+                          left: { metric: "revenue" },
+                          right: { metric: "cost" },
+                        },
+                        right: { number: 100 },
+                      }
+                    : {
+                        operator: "subtract",
+                        left: { metric: "revenue" },
+                        right: { metric: "cost" },
+                      },
               },
             },
     });
@@ -123,11 +134,33 @@ test("新增列：补充信息触发追问并重新确认后生成公式工作
   const reopened = page.getByRole("dialog", { name: "生成公式工作簿" });
   await reopened.getByLabel("已保存定义").selectOption({ label: "收入成本差" });
   await expect(reopened).toContainText("计算表达式：(实际收入 − 成本)");
-  await reopened
-    .getByRole("button", { name: "确认并生成公式工作簿" })
-    .click();
+  // F04 回归：载入定义后，原定义作为可见上下文进入对话
+  await expect(reopened).toContainText("已载入已保存定义「收入成本差」");
+  await expect(reopened).toContainText("我：已载入已保存定义");
+  await expect(reopened).toContainText("助手：已载入该定义作为当前基准");
+  // F04 回归：选择空选项时同步清理当前定义与上下文，选择状态与预览一致
+  await reopened.getByLabel("已保存定义").selectOption({ label: "选择定义" });
+  await expect(reopened.getByText("计算表达式：(实际收入 − 成本)")).toHaveCount(
+    0,
+  );
+  await expect(reopened.getByText("已载入已保存定义")).toHaveCount(0);
+
+  await reopened.getByLabel("已保存定义").selectOption({ label: "收入成本差" });
+  await reopened.getByRole("checkbox").check();
+  await reopened.getByLabel("列计算描述").fill("把刚才公式改成乘以 100");
+  await reopened.getByRole("button", { name: "发送描述" }).click();
+  await expect(reopened).toContainText(
+    "计算表达式：((实际收入 − 成本) × 100)",
+  );
+  // 模型请求必须包含载入的原定义上下文和新的修改描述
+  expect(sentMessages.at(0)?.content).toContain(
+    "已载入已保存定义「收入成本差」",
+  );
+  expect(sentMessages.at(0)?.content).toContain('"operator":"subtract"');
+  expect(sentMessages.at(-1)?.content).toBe("把刚才公式改成乘以 100");
+  await reopened.getByRole("button", { name: "确认并生成公式工作簿" }).click();
   await expect(reopened.getByRole("status")).toContainText("生成完成", {
     timeout: 30000,
   });
-  expect(designCalls).toBe(4);
+  expect(designCalls).toBe(5);
 });

tests/e2e/workspace.spec.ts

@@ -30,7 +30,9 @@ test("桌面：真实导入、异常、下钻、证据、设置、Copilot、导
   });
   await uploaded;
   const dialog = page.getByRole("dialog");
-  await expect(dialog.getByRole("heading", { name: "确认导入字段" })).toBeVisible();
+  await expect(
+    dialog.getByRole("heading", { name: "确认导入字段" }),
+  ).toBeVisible();
   await dialog.getByRole("button", { name: "确认并导入" }).click();
   await expect(
     page.getByRole("heading", { name: "集团经营总览", exact: true }),
@@ -85,14 +87,27 @@ test("桌面：真实导入、异常、下钻、证据、设置、Copilot、导
   await expect(evidenceDialog).toContainText("1 条");
   await expect(evidenceDialog).toContainText("2026-08");
   await expect(evidenceDialog).toContainText("CSV");
-  await evidenceDialog.getByRole("button", { name: "基期", exact: true }).click();
+  await evidenceDialog
+    .getByRole("button", { name: "基期", exact: true })
+    .click();
   await expect(evidenceDialog).toContainText("2025-08");
   await evidenceDialog.getByLabel("关闭", { exact: true }).click();
   const evidenceOffsets: number[] = [];
+  let evidenceFailOnce = false;
+  let evidenceEmpty = false;
   await page.route("**/api/datasets/*/evidence", async (route) => {
     const request = route.request().postDataJSON() as { offset?: number };
     const offset = request.offset || 0;
+    if (evidenceFailOnce) {
+      evidenceFailOnce = false;
+      await route.fulfill({ status: 500, json: { error: "服务暂不可用" } });
+      return;
+    }
     evidenceOffsets.push(offset);
+    if (evidenceEmpty) {
+      await route.fulfill({ json: { total: 0, sheet: "CSV", rows: [] } });
+      return;
+    }
     const count = offset === 0 ? 50 : 1;
     await route.fulfill({
       json: {
@@ -111,10 +126,51 @@ test("桌面：真实导入、异常、下钻、证据、设置、Copilot、导
   });
   await page.getByLabel("查看医生A证据", { exact: true }).click();
   await expect(evidenceDialog).toContainText("第 1-50 条，共 51 条");
-  await evidenceDialog.getByRole("button", { name: "下一页", exact: true }).click();
+  await evidenceDialog
+    .getByRole("button", { name: "下一页", exact: true })
+    .click();
   await expect(evidenceDialog).toContainText("第 51-51 条，共 51 条");
   await expect(evidenceDialog).toContainText("2026-09");
   expect(evidenceOffsets).toEqual([0, 50]);
+  // F03 回归：第二页时重复点击当前期间应回到第一页并重新加载
+  await evidenceDialog
+    .getByRole("button", { name: "本期", exact: true })
+    .click();
+  await expect(evidenceDialog).toContainText("第 1-50 条，共 51 条");
+  expect(evidenceOffsets).toEqual([0, 50, 0]);
+  // F03 回归：第一页重复点击当前期间不清空列表、不重新请求、不停在“加载中”
+  await evidenceDialog
+    .getByRole("button", { name: "本期", exact: true })
+    .click();
+  await expect(evidenceDialog).toContainText("第 1-50 条，共 51 条");
+  await expect(evidenceDialog).not.toContainText("加载中");
+  expect(evidenceOffsets).toEqual([0, 50, 0]);
+  // F03 回归：快速本期/基期切换最终停留在当前期间的数据
+  await evidenceDialog
+    .getByRole("button", { name: "基期", exact: true })
+    .click();
+  await evidenceDialog
+    .getByRole("button", { name: "本期", exact: true })
+    .click();
+  await expect(evidenceDialog).toContainText("第 1-50 条，共 51 条");
+  await expect(evidenceDialog).not.toContainText("加载中");
+  // F03 回归：请求失败显示错误并可重试恢复
+  evidenceFailOnce = true;
+  await evidenceDialog
+    .getByRole("button", { name: "下一页", exact: true })
+    .click();
+  await expect(evidenceDialog).toContainText("服务暂不可用");
+  await expect(evidenceDialog).toContainText("加载失败");
+  await evidenceDialog.getByRole("button", { name: "重试" }).click();
+  await expect(evidenceDialog).toContainText("第 51-51 条，共 51 条");
+  // F03 回归：空结果显示空态而不是过期数据或加载中
+  evidenceEmpty = true;
+  await evidenceDialog
+    .getByRole("button", { name: "上一页", exact: true })
+    .click();
+  await expect(evidenceDialog).toContainText("本期范围内没有数据");
+  await expect(evidenceDialog).toContainText("共 0 条");
+  await expect(evidenceDialog).not.toContainText("加载中");
   await evidenceDialog.getByLabel("关闭", { exact: true }).click();
   await page
     .getByRole("navigation", { name: "分析路径" })
@@ -164,38 +220,53 @@ test("桌面：真实导入、异常、下钻、证据、设置、Copilot、导
   await page.route("**/formula-design", async (route) => {
     designCalls++;
     await route.fulfill({
-      json: designCalls === 1
-        ? { complete: false, questions: ["缺失值和零分母如何处理？"], column: null }
-        : {
-            complete: true,
-            questions: [],
-            column: {
-              name: "收入成本差",
-              meaning: "实际收入减成本",
-              scope: "所选分组与期间",
-              missing: "结果不可用（#N/A）；零分母不可用",
-              expression: {
-                operator: "subtract",
-                left: { metric: "revenue" },
-                right: { metric: "cost" },
+      json:
+        designCalls === 1
+          ? {
+              complete: false,
+              questions: ["缺失值和零分母如何处理？"],
+              column: null,
+            }
+          : {
+              complete: true,
+              questions: [],
+              column: {
+                name: "收入成本差",
+                meaning: "实际收入减成本",
+                scope: "所选分组与期间",
+                missing: "结果不可用（#N/A）；零分母不可用",
+                expression: {
+                  operator: "subtract",
+                  left: { metric: "revenue" },
+                  right: { metric: "cost" },
+                },
               },
             },
-          },
     });
   });
   await page.getByRole("button", { name: "生成公式工作簿" }).click();
   const formulaDialog = page.getByRole("dialog", { name: "生成公式工作簿" });
   await formulaDialog.getByRole("checkbox").check();
-  await formulaDialog.getByLabel("列计算描述").fill("新增收入成本差，缺失或零分母显示不可用");
+  await formulaDialog
+    .getByLabel("列计算描述")
+    .fill("新增收入成本差，缺失或零分母显示不可用");
   await formulaDialog.getByRole("button", { name: "发送描述" }).click();
   await expect(formulaDialog).toContainText("缺失值和零分母如何处理？");
-  await formulaDialog.getByLabel("列计算描述").fill("缺失项和零分母都按不可用处理");
+  await formulaDialog
+    .getByLabel("列计算描述")
+    .fill("缺失项和零分母都按不可用处理");
   await formulaDialog.getByRole("button", { name: "发送描述" }).click();
-  await formulaDialog.getByRole("button", { name: "确认并生成公式工作簿" }).click();
-  await expect(formulaDialog.getByRole("status")).toContainText("生成完成", { timeout: 30000 });
+  await formulaDialog
+    .getByRole("button", { name: "确认并生成公式工作簿" })
+    .click();
+  await expect(formulaDialog.getByRole("status")).toContainText("生成完成", {
+    timeout: 30000,
+  });
   const formulaDownload = page.waitForEvent("download");
   await formulaDialog.getByRole("button", { name: "下载公式工作簿" }).click();
-  expect((await formulaDownload).suggestedFilename()).toBe("traceiq-formula-analysis.xlsx");
+  expect((await formulaDownload).suggestedFilename()).toBe(
+    "traceiq-formula-analysis.xlsx",
+  );
   await formulaDialog.getByLabel("关闭").click();
 });
 test("手机：登录、总览、设置、Copilot 布局无横向溢出", async ({ page }) => {

tests/test_formula_cleanup.py

@@ -1,4 +1,5 @@
 import os
+import uuid
 from datetime import timedelta
 
 import pytest
@@ -27,6 +28,12 @@ def cleanup_root(settings, tmp_path):
     return tmp_path / "formula-exports", dataset, user
 
 
+def _aged(path, hours=25):
+    stamp = (timezone.now() - timedelta(hours=hours)).timestamp()
+    os.utime(path, (stamp, stamp))
+    return path
+
+
 def test_cleanup_expires_jobs_and_removes_files(cleanup_root):
     root, dataset, user = cleanup_root
     path = root / str(user.pk) / "expired.xlsx"
@@ -55,8 +62,7 @@ def test_cleanup_removes_only_stale_temporary_files(cleanup_root):
     stale, recent = user_dir / "stale.tmp", user_dir / "recent.tmp"
     stale.touch()
     recent.touch()
-    old = timezone.now().timestamp() - 25 * 3600
-    os.utime(stale, (old, old))
+    _aged(stale)
 
     call_command("cleanup_formula_exports")
 
@@ -81,3 +87,71 @@ def test_cleanup_does_not_delete_files_outside_export_root(cleanup_root, tmp_pat
     job.refresh_from_db()
     assert job.status == "expired"
     assert outside.read_bytes() == b"keep"
+
+
+def test_cleanup_reclaims_failed_job_leftover_after_retention(cleanup_root):
+    root, dataset, user = cleanup_root
+    job = FormulaExport.objects.create(dataset=dataset, user=user, status="failed")
+    path = root / str(user.pk) / f"{job.pk}.xlsx"
+    path.parent.mkdir(parents=True)
+    path.write_bytes(b"leftover")
+    job.path = str(path)
+    job.save(update_fields=["path"])
+    _aged(path)
+
+    call_command("cleanup_formula_exports")
+
+    job.refresh_from_db()
+    assert not path.exists()
+    assert job.path == ""
+
+
+def test_cleanup_keeps_workbook_of_unexpired_succeeded_job(cleanup_root):
+    root, dataset, user = cleanup_root
+    job = FormulaExport.objects.create(
+        dataset=dataset,
+        user=user,
+        status="succeeded",
+        expires_at=timezone.now() + timedelta(hours=1),
+    )
+    path = root / str(user.pk) / f"{job.pk}.xlsx"
+    path.parent.mkdir(parents=True)
+    path.write_bytes(b"active")
+    job.path = str(path)
+    job.save(update_fields=["path"])
+    _aged(path)
+
+    call_command("cleanup_formula_exports")
+
+    assert path.read_bytes() == b"active"
+    job.refresh_from_db()
+    assert job.path == str(path)
+
+
+def test_cleanup_reclaims_orphan_whose_job_row_was_deleted(cleanup_root):
+    root, _, user = cleanup_root
+    user_dir = root / str(user.pk)
+    user_dir.mkdir(parents=True)
+    orphan = user_dir / f"{uuid.uuid4()}.xlsx"
+    orphan.write_bytes(b"orphan")
+    recent = user_dir / f"{uuid.uuid4()}.xlsx"
+    recent.write_bytes(b"recent orphan")
+    _aged(orphan)
+
+    call_command("cleanup_formula_exports")
+
+    assert not orphan.exists()
+    assert recent.exists()
+
+
+def test_cleanup_ignores_non_uuid_workbook_names(cleanup_root):
+    root, _, user = cleanup_root
+    user_dir = root / str(user.pk)
+    user_dir.mkdir(parents=True)
+    odd = user_dir / "notes.xlsx"
+    odd.write_bytes(b"keep")
+    _aged(odd)
+
+    call_command("cleanup_formula_exports")
+
+    assert odd.exists()

tests/test_formula_columns.py

@@ -200,6 +200,54 @@ def test_design_column_preserves_clarification_branch_and_full_conversation(form
     assert payload["input"][1:] == messages
 
 
+def test_design_column_accepts_loaded_definition_context(formula_data, settings):
+    ds, _ = formula_data
+    settings.AI_ENABLED = True
+    settings.OPENAI_API_KEY = "test-key"
+    settings.OPENAI_MODEL = "test-model"
+    context = "已载入已保存定义「收入成本差」；计算表达式：" + '{"operator":"subtract",' * 45
+    assert 1000 < len(context) <= 2000
+    messages = [
+        {"role": "user", "content": context},
+        {"role": "assistant", "content": "已载入该定义作为当前基准。"},
+        {"role": "user", "content": "把刚才公式改成乘以 100"},
+    ]
+    clarification = {
+        "status": "completed",
+        "output": [
+            {
+                "content": [
+                    {
+                        "type": "output_text",
+                        "text": '{"complete":false,"questions":["要乘在整个表达式上吗？"],"column":null}',
+                    }
+                ]
+            }
+        ],
+    }
+    with patch("finance.formula_views.httpx.Client") as http:
+        http.return_value.__enter__.return_value.post.return_value = type(
+            "Response",
+            (),
+            {"raise_for_status": lambda self: None, "json": lambda self: clarification},
+        )()
+        result = design_column(
+            messages,
+            ds,
+            {"current": ["2026-08"], "baseline": ["2025-08"], "filters": {}, "group": "department"},
+        )
+    assert result["complete"] is False
+    payload = http.return_value.__enter__.return_value.post.call_args.kwargs["json"]
+    assert payload["input"][1:] == messages
+
+    with pytest.raises(Exception, match="对话消息内容无效"):
+        design_column(
+            [{"role": "user", "content": "y" * 2001}],
+            ds,
+            {"current": ["2026-08"], "baseline": ["2025-08"], "filters": {}, "group": "department"},
+        )
+
+
 def test_design_column_returns_only_validated_formula_definition(formula_data, settings):
     ds, _ = formula_data
     settings.AI_ENABLED = True

tests/test_formula_export_reliability.py

@@ -1,3 +1,4 @@
+import os
 import stat
 from decimal import Decimal
 from io import BytesIO
@@ -198,6 +199,170 @@ def test_formula_export_worker_records_path_preparation_failure(workspace, setti
     assert job.status == "failed"
 
 
+def _queued_job(ds):
+    return FormulaExport.objects.create(
+        dataset=ds,
+        user=ds.created_by,
+        context={"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
+    )
+
+
+def test_failed_export_rejects_symlinked_user_directory_without_touching_files(workspace, settings):
+    from finance.exporting import process_formula_export
+
+    ds, *_ = workspace
+    job = _queued_job(ds)
+    root = Path(settings.MEDIA_ROOT) / "formula-exports"
+    root.mkdir(parents=True)
+    protected = Path(settings.MEDIA_ROOT) / "protected"
+    protected.mkdir()
+    victim = protected / f"{job.pk}.xlsx"
+    victim.write_bytes(b"protected workbook")
+    (root / str(job.user_id)).symlink_to(protected, target_is_directory=True)
+
+    with pytest.raises(ValueError, match="符号链接"):
+        process_formula_export(job.pk)
+
+    job.refresh_from_db()
+    assert job.status == "failed"
+    assert not job.path
+    assert victim.read_bytes() == b"protected workbook"
+    assert list(protected.glob("*.tmp")) == []
+
+
+def test_failed_export_rejects_symlinked_target_and_keeps_link(workspace, settings):
+    from finance.exporting import process_formula_export
+
+    ds, *_ = workspace
+    job = _queued_job(ds)
+    user_dir = Path(settings.MEDIA_ROOT) / "formula-exports" / str(job.user_id)
+    user_dir.mkdir(parents=True)
+    protected_file = Path(settings.MEDIA_ROOT) / "protected.xlsx"
+    protected_file.write_bytes(b"protected workbook")
+    (user_dir / f"{job.pk}.xlsx").symlink_to(protected_file)
+
+    with pytest.raises(ValueError, match="普通文件"):
+        process_formula_export(job.pk)
+
+    job.refresh_from_db()
+    assert job.status == "failed"
+    assert protected_file.read_bytes() == b"protected workbook"
+    assert (user_dir / f"{job.pk}.xlsx").is_symlink()
+    assert list(user_dir.glob("*.tmp")) == []
+
+
+def test_failed_export_refuses_cleanup_when_parent_directory_swapped_to_symlink(
+    workspace, settings, monkeypatch
+):
+    from finance.exporting import process_formula_export
+
+    ds, *_ = workspace
+    job = _queued_job(ds)
+    root = Path(settings.MEDIA_ROOT) / "formula-exports"
+    user_dir = root / str(job.user_id)
+    user_dir.mkdir(parents=True)
+    protected = Path(settings.MEDIA_ROOT) / "protected"
+    protected.mkdir()
+    victim = protected / f"{job.pk}.xlsx"
+    victim.write_bytes(b"protected workbook")
+
+    def swap_and_fail(*args, **kwargs):
+        args[-1].write(b"partial")
+        user_dir.rename(user_dir.with_name("swapped"))
+        user_dir.symlink_to(protected, target_is_directory=True)
+        raise ValueError("controlled failure after path validation")
+
+    monkeypatch.setattr("finance.exporting.build_formula_workbook", swap_and_fail)
+    with pytest.raises(ValueError, match="controlled failure"):
+        process_formula_export(job.pk)
+
+    job.refresh_from_db()
+    assert job.status == "failed"
+    assert victim.read_bytes() == b"protected workbook"
+    assert user_dir.is_symlink()
+    # 本次创建的临时文件残留在被换名的原目录中，由清理命令按保留期回收。
+    assert list((root / "swapped").glob(f"{job.pk}.*.tmp"))
+
+
+def test_formula_export_retry_after_failure_succeeds(workspace, monkeypatch):
+    from finance.exporting import process_formula_export
+
+    ds, *_ = workspace
+    job = _queued_job(ds)
+
+    def fail(*args, **kwargs):
+        args[-1].write(b"partial")
+        raise ValueError("controlled export failure")
+
+    monkeypatch.setattr("finance.exporting.build_formula_workbook", fail)
+    with pytest.raises(ValueError, match="controlled export failure"):
+        process_formula_export(job.pk)
+    monkeypatch.undo()
+
+    stats = process_formula_export(job.pk)
+    job.refresh_from_db()
+    assert stats["rows"] == 5
+    assert job.status == "succeeded"
+    assert Path(job.path).read_bytes()[:2] == b"PK"
+    assert not list(Path(job.path).parent.glob(f"{job.pk}.*.tmp"))
+
+
+def test_cleanup_failure_is_logged_and_reclaimed_after_permissions_restored(
+    workspace, settings, monkeypatch, caplog
+):
+    import logging as logging_module
+    from datetime import timedelta
+
+    from django.core.management import call_command
+    from django.utils import timezone
+
+    from finance.exporting import process_formula_export
+
+    ds, *_ = workspace
+    job = _queued_job(ds)
+    user_dir = Path(settings.MEDIA_ROOT) / "formula-exports" / str(job.user_id)
+    user_dir.mkdir(parents=True)
+    target = user_dir / f"{job.pk}.xlsx"
+    target.write_bytes(b"stale workbook")
+    other = FormulaExport.objects.create(
+        dataset=ds,
+        user=ds.created_by,
+        context={"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
+        status="succeeded",
+        expires_at=timezone.now() + timedelta(hours=1),
+    )
+    kept = user_dir / f"{other.pk}.xlsx"
+    other.path = str(kept)
+    other.save(update_fields=["path"])
+    kept.write_bytes(b"active workbook")
+
+    def fail(*args, **kwargs):
+        args[-1].write(b"partial")
+        os.chmod(user_dir, 0o500)
+        raise ValueError("controlled export failure")
+
+    monkeypatch.setattr("finance.exporting.build_formula_workbook", fail)
+    with pytest.raises(ValueError, match="controlled export failure"):
+        process_formula_export(job.pk)
+    job.refresh_from_db()
+    assert job.status == "failed"
+    assert job.path == str(target)
+    warnings = [r for r in caplog.records if r.levelno == logging_module.WARNING]
+    assert warnings and all(str(job.pk) in record.getMessage() for record in warnings)
+
+    os.chmod(user_dir, 0o700)
+    old = (timezone.now() - timedelta(hours=25)).timestamp()
+    for leftover in [target, *user_dir.glob(f"{job.pk}.*.tmp"), kept]:
+        os.utime(leftover, (old, old))
+    call_command("cleanup_formula_exports")
+
+    job.refresh_from_db()
+    assert not target.exists()
+    assert not list(user_dir.glob(f"{job.pk}.*.tmp"))
+    assert job.path == ""
+    assert kept.read_bytes() == b"active workbook"
+
+
 def test_formula_workbook_rejects_unclassified_label_collision(workspace):
     from decimal import Decimal
 

tests/test_worker_paths.py

@@ -1,9 +1,12 @@
+from datetime import timedelta
+from pathlib import Path
 from unittest.mock import patch
 
 import pytest
 from django.core.management import call_command
+from django.utils import timezone
 
-from finance.models import FormulaExport
+from finance.models import Dataset, FormulaExport
 
 pytestmark = pytest.mark.django_db
 
@@ -17,6 +20,10 @@ def test_worker_processes_queued_import_and_reports_export_failure(workspace):
     ) as process_import:
         call_command("import_worker", once=True)
     process_import.assert_called_once_with(ds.pk)
+    # PostgreSQL 上 worker 持咨询锁会立即重认领 processing 任务：模拟导入完成，
+    # 否则第二次 --once 又被它抢占，轮不到下面的导出任务（SQLite 无咨询锁，
+    # 不走立即恢复路径，两种后端行为一致）。
+    Dataset.objects.filter(pk=ds.pk).update(status="ready")
 
     export = FormulaExport.objects.create(dataset=ds, user=ds.created_by, context={})
     with patch(
@@ -26,3 +33,118 @@ def test_worker_processes_queued_import_and_reports_export_failure(workspace):
         call_command("import_worker", once=True)
     export.refresh_from_db()
     assert export.status == "processing"
+
+
+def _export_job(ds, **kwargs):
+    return FormulaExport.objects.create(
+        dataset=ds,
+        user=ds.created_by,
+        context={"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
+        **kwargs,
+    )
+
+
+def test_worker_refuses_export_after_role_revocation(workspace):
+    from finance.exporting import process_formula_export
+
+    ds, member, *_ = workspace
+    job = _export_job(ds, status="processing")
+    member.role = "viewer"
+    member.save(update_fields=["role"])
+
+    with pytest.raises(PermissionError, match="角色"):
+        process_formula_export(job.pk)
+
+    job.refresh_from_db()
+    assert job.status == "failed"
+    assert not job.path
+    assert not list(Path(ds.file.storage.location).rglob(f"{job.pk}*"))
+
+
+def test_worker_refuses_export_after_account_deactivation(workspace):
+    from finance.exporting import process_formula_export
+
+    ds, _, _, _, _ = workspace
+    user = ds.created_by
+    user.is_active = False
+    user.save(update_fields=["is_active"])
+    job = _export_job(ds, status="processing")
+
+    with pytest.raises(PermissionError, match="停用"):
+        process_formula_export(job.pk)
+
+    job.refresh_from_db()
+    assert job.status == "failed"
+
+
+def test_status_and_download_deny_revoked_viewer(workspace, settings):
+    ds, member, client, *_ = workspace
+    job = _export_job(
+        ds,
+        status="succeeded",
+        finished_at=timezone.now(),
+        expires_at=timezone.now() + timedelta(hours=1),
+    )
+    target = Path(settings.MEDIA_ROOT) / "formula-exports" / str(job.user_id) / f"{job.pk}.xlsx"
+    target.parent.mkdir(parents=True)
+    target.write_bytes(b"PK workbook")
+    job.path = str(target)
+    job.save(update_fields=["path"])
+
+    assert client.get(f"/api/datasets/{ds.pk}/formula-exports/{job.pk}").status_code == 200
+    assert (
+        client.get(f"/api/datasets/{ds.pk}/formula-exports/{job.pk}/download").status_code == 200
+    )
+
+    member.role = "viewer"
+    member.save(update_fields=["role"])
+    assert client.get(f"/api/datasets/{ds.pk}/formula-exports/{job.pk}").status_code == 403
+    assert (
+        client.get(f"/api/datasets/{ds.pk}/formula-exports/{job.pk}/download").status_code == 403
+    )
+
+
+def test_half_finished_export_cannot_be_downloaded(workspace, settings):
+    ds, _, client, *_ = workspace
+    for status, code in (("processing", 409), ("failed", 409), ("queued", 409)):
+        job = _export_job(ds, status=status)
+        assert (
+            client.get(f"/api/datasets/{ds.pk}/formula-exports/{job.pk}/download").status_code
+            == code
+        ), status
+    succeeded = _export_job(
+        ds,
+        status="succeeded",
+        path=str(Path(settings.MEDIA_ROOT) / "nowhere.xlsx"),
+        finished_at=timezone.now(),
+        expires_at=timezone.now() + timedelta(hours=1),
+    )
+    assert (
+        client.get(f"/api/datasets/{ds.pk}/formula-exports/{succeeded.pk}/download").status_code
+        == 410
+    )
+
+
+def test_stale_processing_export_is_reclaimed_after_timeout(workspace):
+    ds, *_ = workspace
+    job = _export_job(ds, status="processing")
+    FormulaExport.objects.filter(pk=job.pk).update(
+        created_at=timezone.now() - timedelta(minutes=20)
+    )
+
+    call_command("import_worker", once=True)
+
+    job.refresh_from_db()
+    assert job.status == "succeeded"
+    assert Path(job.path).read_bytes()[:2] == b"PK"
+    assert not list(Path(job.path).parent.glob(f"{job.pk}.*.tmp"))
+
+
+def test_recent_processing_export_is_not_reclaimed(workspace):
+    ds, *_ = workspace
+    job = _export_job(ds, status="processing")
+
+    call_command("import_worker", once=True)
+
+    job.refresh_from_db()
+    assert job.status == "processing"