merge: 同步 main 的 CI 基建（.gitea/workflows 迁移、PG 咨询锁测试适配、artifact 容忍、audit 重试）

runner 到 registry.npmjs.org 的 TLS 连接偶发被断（audit endpoint returned an
error），上一轮测试全绿仍被 audit 步骤标红；改为最多重试 3 次、间隔 15 秒。
真实漏洞是确定性的，重试后依然失败，不会漏报

.gitea/ai-review.md

@@ -0,0 +1,27 @@
+# fa-agent (TraceIQ 财务分析) review profile
+
+## Stack and verification
+
+- Django 5.2 后端（PostgreSQL 生产 / SQLite 仅开发），金额一律 `NUMERIC(44,24)` + Python `Decimal`（精度 60）；JSON 中金额为字符串。
+- React 19 + TypeScript + Vite 前端（AG Grid、ECharts），`npm run build` 含 `tsc --noEmit`。
+- openpyxl 流式生成公式工作簿（SUMIFS/COUNTIFS），公式正确性靠 LibreOffice 无头重算验证。
+- pytest（`--cov-fail-under=85`）+ Playwright E2E；CI 在 PostgreSQL 17 服务容器上跑全量。
+- 验证命令：`uv run pytest`、`uv run ruff check server finance tests scripts`、`npm run typecheck`、`python manage.py makemigrations --check`。
+
+## Review priorities
+
+1. **金额精度**：财务计算、持久化、Excel 公式生成路径上出现 `float()`、`round()`、Decimal 与 float 混算、或金额以数字（非字符串）进 JSON，必须报。浏览器 `Number` 只允许用于绘图与显示。
+2. **缺失值语义**：成本/毛利/人数等依赖字段缺失时必须传播为不可用（服务端 `None`、Excel `#N/A`），不得静默转 0 或跳过；比率分母为零/非正同样不可用。对账容差按指标类型独立（人数严格相等、金额 0.01、比率 0.0001），不得为消除报错统一放宽容差。
+3. **文件与路径安全**：清理/下载/回收文件前必须 `lstat` 校验且拒绝符号链接（目录用 `O_NOFOLLOW` 打开），只删除本次任务在受控目录内创建的文件；清理失败要留日志与可追踪状态，不得静默吞掉 `OSError`。
+4. **越权与生命周期**：每个 endpoint 必须声明 roles 白名单；worker 执行阶段要重新核验角色、`is_active`、组织与科室范围（不能只信创建时的检查）；卡死任务要有超时恢复，且恢复不得与活跃 worker 抢任务。
+5. **事务与并发**：配额/唯一性检查必须在事务内并锁定稳定的所属行（如 Membership），审计与业务变更同事务提交；乐观并发用 revision + 409，不得静默后写覆盖。
+6. **Excel 公式注入与上限**：不可信文本写入单元格必须强制文本类型；CSV/公式中 `= + - @` 开头加 `'` 前缀；SUMIFS/COUNTIFS criteria 必须转义 `~ * ? "` 并检查 255 字符上限、公式 8192 字符上限。
+7. **前端异步状态**：useEffect 依赖未变时不得清空已加载数据（同状态重复点击不能停在“加载中”）；并发请求要有过期响应丢弃（active flag/AbortController）；失败路径要有可见错误与重试。
+8. **测试真实性**：行为修复必须带“修复前失败、修复后通过”的回归测试；环境门禁（LibreOffice、PostgreSQL 并发、真实 AI）必须显式 skip 并注明原因，不得用 mock 通过冒充真实验收。
+
+## Avoid reporting
+
+- 纯格式与风格问题（ruff / prettier 已在 CI 把关）。
+- 改动行为已被等价集成测试或 E2E 覆盖时，仅因缺少同名单测而报缺测试。
+- 没有具体并发路径的假想竞态（SQLite 单写者语义下的理论交错不算，除非改动显式依赖 PostgreSQL 行为）。
+- 对 `docs/` 报告类 HTML/Markdown 的措辞与结构建议——它们记录的是验收状态，不是产品代码。

.gitea/workflows/ai-review.yml

@@ -0,0 +1,202 @@
+name: AI Review
+
+# 对抗式 PR 审查：先一遍找出候选缺陷，再对每一条派一个反驳者去证伪它，只有
+# 活下来的才会发到 PR 上。这个不对称是刻意的——AI 审查员死于误报远早于死于漏
+# 报，所以反驳阶段默认站在"这是误报"一边，"什么都没发现"是可接受的结果。
+#
+# 引擎不在本仓库，在 vertai-tech/ai-review-engine。本文件只负责把三样东西摆到
+# 位再调它：被审查的代码、本仓库的审查侧重、引擎本身。
+#
+# 接入一个仓库需要（各一次）：
+#   1. 引擎仓库的 templates/ai-review.yml -> 本仓库 .gitea/workflows/ai-review.yml
+#   2. 引擎仓库的 templates/ai-review.md  -> 本仓库 .gitea/ai-review.md，按本仓库
+#      的技术栈改写
+#   3. 下面这些组织级机密 / 变量（组织设置 → Actions）配好之后，新仓库就不用再
+#      单独配任何东西：
+#        secret   REVIEW_API_KEY        模型网关的 key
+#        secret   REVIEW_ENGINE_TOKEN   只读 PAT，用于拉私有的引擎仓库
+#        variable REVIEW_BASE_URL       任意 OpenAI 格式的根地址
+#        variable REVIEW_MODEL          支持 function calling 的模型
+#
+# 自动修复默认是关的。要开，在**本仓库**的 设置 → Actions → 变量 里加一个（这两条
+# 都不要配到组织级，那样等于给所有仓库都开了），并在组织级配一次
+# REVIEW_FIX_MODEL：
+#   REVIEW_AUTOFIX=1            改代码、提交、推回 PR 分支
+#   REVIEW_FIX_SUGGEST_ONLY=1   只把改法写进评论，一个字节都不落盘
+# 两个都配的时候，不动仓库的那个赢。fork 来的 PR 推不回去，apply 会自动降级成
+# 只给改法。
+#
+# 默认只修一轮：修完推一个 commit，那次推送起的下一轮只审不改。要"一直修到没
+# 意见"，把仓库变量 REVIEW_FIX_ROUNDS 调大——它数的是这个分支已经被自动修过几
+# 轮，轮次写在提交消息里。
+#
+# 改法本身是两步定出来的：一个模型决定这条该不该由机器改、以及改完长什么样，另一
+# 个模型把它落成逐字命中的替换。想让这两步用不同的模型，就再配一个组织级的
+# REVIEW_FIX_PLAN_MODEL——判断那步值得用贵的，抄写那步不值得。不配也照样分两步，
+# 只是两步都用 REVIEW_FIX_MODEL。
+#
+# 成本由模型决定而不是由这个设计决定：同一个 PR 在便宜模型上是几分钱，在旗舰
+# 模型上是几美元。换模型前先看一眼那家的每 token 价格。
+# 换了模型记得同步组织变量 REVIEW_PRICES，缺单价的那轮评论不报金额。
+#
+# 只发 COMMENT 类型的 review，从不 approve，也从不 request changes。所有发现集中
+# 在一条评论里按 1、2、3 编号，不再逐条钉成行内评论。
+
+on:
+  pull_request:
+    # 没有 ready_for_review：Gitea Actions 不实现这个活动类型（它支持的是
+    # opened / edited / closed / reopened / assigned / unassigned /
+    # synchronize / labeled / unlabeled / milestoned / demilestoned /
+    # review_requested / review_request_removed）。写上去只会让人以为有这个
+    # 覆盖，实际没有。
+    #
+    # 留下的缺口：草稿 PR 会被下面的 draft == false 挡掉，所以"取消草稿但不再
+    # 推代码"不会触发任何东西。再推一次，或者用下面的 workflow_dispatch 手动
+    # 指定 PR 号——那条路就是为这个场景留的。
+    #
+    # 用 edited 能覆盖取消草稿，但它同时会在每次改标题、改描述时触发，而每一
+    # 次都是一整轮 find + refute 的钱。为一个"再推一次就解决"的场景不值。
+    types: [opened, synchronize, reopened]
+  workflow_dispatch:
+    inputs:
+      pr_number:
+        description: "PR number to review"
+        required: true
+        type: string
+
+# 连着推两次的 PR 只该被审一次——取消在飞的那轮，而不是花钱买一份关于旧代码
+# 的报告。
+concurrency:
+  group: ai-review-${{ github.event.pull_request.number || inputs.pr_number }}
+  cancel-in-progress: true
+
+jobs:
+  review:
+    # 草稿按定义就是没写完的东西；手动 dispatch 是明确的请求，所以绕过这条。
+    # vars 那两个条件意味着本文件在网关配好之前合入也是安全的——job 直接跳过。
+    if: >-
+      (github.event_name == 'workflow_dispatch' || github.event.pull_request.draft == false)
+      && vars.REVIEW_BASE_URL != ''
+      && vars.REVIEW_MODEL != ''
+    runs-on: ubuntu-latest
+    # 不设的话是 Gitea 默认的 6 小时，而下面的 continue-on-error 帮不上忙——它
+    # 兜的是失败，不是挂起——于是一次卡在不返回的 API 调用上的审查会一直占着
+    # runner，PR 的 check 一下午都是 pending。fantap-server 真踩过这个。
+    #
+    # 60 分钟是照实测定的：一次阶段一实测 710 秒、29k 输出 token（其中 28.7k
+    # 是推理），阶段二还要在这之上再加最多八条反驳，开了自动修复的话阶段三再加
+    # 最多八条的规划加执行。阶段三从每条一次调用变成每条两次（执行没对上文件还
+    # 会重试）之后，原来的 45 分钟只在八条全中时才不够，但不够的表现是卡在超时
+    # 上而不是干净报错，所以往上留一档。上面 concurrency 的 cancel-in-progress
+    # 负责拦住堆积。
+    timeout-minutes: 60
+    permissions:
+      # write 只有 REVIEW_AUTOFIX 那个模式才用得上——它要把改动推回 PR 分支。
+      # 没开自动修复、或者只开了 REVIEW_FIX_SUGGEST_ONLY 的仓库，把这一行改回
+      # read 就是了：只给改法的模式一个字节都不落盘，不需要写权限。留着 write
+      # 也没事，引擎只在 apply 模式下才提交任何东西。
+      contents: write
+      pull-requests: write
+    steps:
+      # 一、被审查的代码。反驳阶段要读 PR head 上的整份文件，所以检出 head，
+      # 不是 merge commit。
+      - uses: actions/checkout@v4
+        with:
+          ref: ${{ github.event.pull_request.head.sha || format('refs/pull/{0}/head', inputs.pr_number) }}
+
+      # 二、本仓库的审查侧重（.gitea/ai-review.md），取默认分支的那一份，不取
+      # PR 分支的——否则一个 PR 可以先改掉审查重点，再让改过的重点来审自己。
+      # || 'main' 是给 Gitea 兜底：它的 workflow_dispatch 载荷不总是带
+      # github.event.repository。
+      - uses: actions/checkout@v4
+        with:
+          ref: ${{ github.event.repository.default_branch || 'main' }}
+          sparse-checkout: .gitea
+          path: .review-profile
+
+      # 三、引擎。跟着 v1 这个移动标签走：引擎修了 bug 移一次标签，所有接入的
+      # 仓库同时生效。某个仓库被新版引擎坑到时，把这里单独钉到具体 commit 就能
+      # 脱身，不用回滚全局。
+      #
+      # 引擎仓库是私有的，Gitea 的 GITHUB_TOKEN 只对本仓库有效，拉不动它，所以
+      # 要一个组织级的只读 PAT。continue-on-error 在这里是必要的：PAT 没配或者
+      # 过期时，这一步失败不该把别人的 PR 弄红——下面那步会说明情况然后干净退出。
+      - uses: actions/checkout@v4
+        continue-on-error: true
+        with:
+          repository: vertai-tech/ai-review-engine
+          ref: v1
+          path: .review-engine
+          token: ${{ secrets.REVIEW_ENGINE_TOKEN }}
+
+      - name: Adversarial review
+        # 设计上就是建议性的：引擎自己的错误已经会干净退出，这里兜的是剩下的
+        # （进程被杀、脚本没看见的 curl 失败）。审查机器人不该让人的 PR 变红。
+        continue-on-error: true
+        # 比 job 的预算小一档，这样超时是记在这一步上，而不是整个 job 被砍——
+        # 后者在界面上看不出是谁超的时。引擎内部 curl 的 max-time 比这再小一档
+        # （1200s），所以正常顺序是那一层带着响应体和日志先报错，而不是这一层
+        # 无声地杀掉进程。
+        timeout-minutes: 55
+        env:
+          REVIEW_API_KEY: ${{ secrets.REVIEW_API_KEY }}
+          REVIEW_BASE_URL: ${{ vars.REVIEW_BASE_URL }}
+          REVIEW_MODEL: ${{ vars.REVIEW_MODEL }}
+          # 可选，组织级变量。网关不在响应里报金额时（sub2api 就不报），评论按这张
+          # 单价表算金额；不配就只报 token。格式见引擎 README「每轮花了多少」。
+          REVIEW_PRICES: ${{ vars.REVIEW_PRICES }}
+          # 上游报过载 / 限流时每次重试前等几秒。不配就是 10,30,60，写 0 关掉。
+          REVIEW_BUSY_RETRY_WAITS: ${{ vars.REVIEW_BUSY_RETRY_WAITS }}
+          # 评论里行号链接用的站点地址，如 https://git.sciark.cn。runner 在容器里访问
+          # Gitea 时，API 返回的 html_url 和 GITHUB_SERVER_URL 都可能是容器内地址，
+          # 不配这个链接就打不开。组织级配一次。
+          REVIEW_PUBLIC_URL: ${{ vars.REVIEW_PUBLIC_URL }}
+          # PR 号和仓库名经 env 进脚本，不直接拼进 run：手动 dispatch 的 pr_number
+          # 是任意字符串，表达式在 shell 解析之前就被展开，带引号就能注入命令。
+          PR_NUMBER: ${{ github.event.pull_request.number || inputs.pr_number }}
+          REVIEW_REPO: ${{ github.repository }}
+          # 引擎的 Gitea REST helper 读这个。GITHUB_API_URL 由 act_runner 导出，
+          # 已经指向 <instance>/api/v1。
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          # 自动修复。REVIEW_AUTOFIX 没配或不是 1 的话，下面几个都没有意义。
+          REVIEW_AUTOFIX: ${{ vars.REVIEW_AUTOFIX }}
+          REVIEW_FIX_SUGGEST_ONLY: ${{ vars.REVIEW_FIX_SUGGEST_ONLY }}
+          REVIEW_FIX_MODEL: ${{ vars.REVIEW_FIX_MODEL }}
+          # 一个分支最多自动修复几轮。不配就是 1：修一次，下一轮只审，跟从前
+          # 一样。配大了才是"一直修到没意见"，代价是往作者分支上推更多 commit、
+          # 一个 PR 烧更多轮的钱。硬上限在引擎里。
+          REVIEW_FIX_ROUNDS: ${{ vars.REVIEW_FIX_ROUNDS }}
+          # 规划用哪个模型。这不是开关——不配也照样先规划再执行，只是两步同一个
+          # 模型。配了才是"贵的判断、便宜的抄写"那种分工。
+          REVIEW_FIX_PLAN_MODEL: ${{ vars.REVIEW_FIX_PLAN_MODEL }}
+          # 只有当写修复的模型不在审查那个网关上时，才需要这两个。
+          REVIEW_FIX_BASE_URL: ${{ vars.REVIEW_FIX_BASE_URL }}
+          REVIEW_FIX_API_KEY: ${{ secrets.REVIEW_FIX_API_KEY }}
+          # 同上，只有当规划的模型连修复那个网关都不在时才需要。
+          REVIEW_FIX_PLAN_BASE_URL: ${{ vars.REVIEW_FIX_PLAN_BASE_URL }}
+          REVIEW_FIX_PLAN_API_KEY: ${{ secrets.REVIEW_FIX_PLAN_API_KEY }}
+          # 规划的思考强度。不配就是模型默认；写 high 之类的词按 reasoning_effort
+          # 发，写 JSON 对象就原样并进请求。各家写法见引擎 README。
+          REVIEW_FIX_PLAN_THINKING: ${{ vars.REVIEW_FIX_PLAN_THINKING }}
+          # 对照用的第二个执行模型。只在直接改的模式下跑，改动不落盘，只和
+          # REVIEW_FIX_MODEL 的改动一起贴进评论，方便比较哪个模型抄得好。
+          REVIEW_FIX_MODEL2: ${{ vars.REVIEW_FIX_MODEL2 }}
+          REVIEW_FIX_MODEL2_BASE_URL: ${{ vars.REVIEW_FIX_MODEL2_BASE_URL }}
+          REVIEW_FIX_MODEL2_API_KEY: ${{ secrets.REVIEW_FIX_MODEL2_API_KEY }}
+          # 只有当上面那个自动 token 推不动分支时，才需要这个写权限 PAT。
+          REVIEW_FIX_TOKEN: ${{ secrets.REVIEW_FIX_TOKEN }}
+        run: |
+          ENGINE=.review-engine/scripts/ai_review.py
+          if [ ! -f "$ENGINE" ]; then
+            # 拉引擎那步失败了。说清楚原因，否则这里的沉默会被读成"没发现问题"。
+            echo "::warning::引擎没拉下来（多半是组织级 secret REVIEW_ENGINE_TOKEN 没配或已过期），本轮未审查。"
+            exit 0
+          fi
+
+          case "$PR_NUMBER" in
+            ''|*[!0-9]*)
+              echo "::warning::PR 号必须是数字，收到的是：$PR_NUMBER。本轮未审查。"
+              exit 0 ;;
+          esac
+          python3 "$ENGINE" --repo "$REVIEW_REPO" --pr "$PR_NUMBER"
+

.gitea/workflows/ci.yml

@@ -16,7 +16,9 @@ jobs:
           --health-cmd pg_isready --health-interval 5s
           --health-timeout 5s --health-retries 10
     env:
-      DATABASE_URL: postgresql://postgres:ci-only-password@localhost:5432/traceiq
+      # Gitea act_runner 的 job 跑在容器里，与服务容器同网络：主机名必须用服务名
+      # postgres（GitHub 托管 runner 的 localhost 语义在这里连不通，勿改回）。
+      DATABASE_URL: postgresql://postgres:ci-only-password@postgres:5432/traceiq
       E2E_PASSWORD: CI-Test-Only-SecurePassword882!
       PYTHONPATH: .
     steps:
@@ -35,8 +37,24 @@ jobs:
       - run: uv run python manage.py makemigrations --check --dry-run
       - run: uv run pytest --cov=finance --cov-report=term-missing --cov-fail-under=85
       - run: npm run build
-      - run: npm audit --omit=dev --registry=https://registry.npmjs.org
-      - run: uv run pip-audit
+      # 审计端点在 runner 网络下偶发 TLS 断开（npm error audit endpoint returned
+      # an error），重试区分网络抖动与真实漏洞：漏洞是确定性的，重试后照样红。
+      - name: npm audit (production deps)
+        run: |
+          for attempt in 1 2 3; do
+            npm audit --omit=dev --registry=https://registry.npmjs.org && exit 0
+            echo "::warning::npm audit 第 ${attempt} 次失败，15 秒后重试"
+            sleep 15
+          done
+          exit 1
+      - name: pip-audit
+        run: |
+          for attempt in 1 2 3; do
+            uv run pip-audit && exit 0
+            echo "::warning::pip-audit 第 ${attempt} 次失败，15 秒后重试"
+            sleep 15
+          done
+          exit 1
       - run: uv run python manage.py migrate --noinput
       - run: uv run python manage.py collectstatic --noinput
       - run: uv run python manage.py bootstrap --username admin --organization CI
@@ -53,8 +71,11 @@ jobs:
           trap 'kill "$web" "$worker"' EXIT
           for i in {1..30}; do curl -fsS http://127.0.0.1:8000/api/health && break; sleep 1; done
           npm run test:e2e
+      # Gitea 不支持 upload-artifact@v4 的 artifact 协议（GHESNotSupportedError），
+      # 上传失败不该把已经全绿的测试标红；迁到 GitHub 托管时此步骤自动生效。
       - uses: actions/upload-artifact@v4
         if: always()
+        continue-on-error: true
         with:
           name: browser-results
           path: |

README.md

@@ -33,6 +33,14 @@
 
 需要 Node.js 22、uv、Python 3.12、PostgreSQL 17。所有示例从仓库根目录执行。
 
+快速启动本地页面（迁移、管理员初始化、前端构建、API 与导入 worker 一条命令完成，健康检查通过后自动打开 `http://127.0.0.1:8000`；`Ctrl-C` 或 `scripts/dev_up.sh --stop` 停止，`--no-open` 跳过自动打开浏览器）：
+
+```bash
+scripts/dev_up.sh
+```
+
+未设置 `DATABASE_URL` 时该脚本使用 SQLite 轻量开发库 `var/db.sqlite3`，仅限页面测试与开发回归；金额聚合与财务验收需按下方配置 PostgreSQL。手动分步流程如下：
+
 ```bash
 uv sync --frozen --python 3.12
 npm ci --registry=https://registry.npmjs.org
@@ -99,7 +107,7 @@ npm audit --omit=dev --registry=https://registry.npmjs.org
 uv run pip-audit
 ```
 
-Playwright 需要正在运行的本地 API 和 worker，可用 E2E_PASSWORD 指定测试 admin 密码，默认读取本地凭据文件。首次执行 `npx playwright install chromium`。浏览器 fixture 测试会新建 `e2e-finance.csv` 数据集，不可指向正式生产环境。`.github/workflows/ci.yml` 在独立 PostgreSQL 上执行同样流程，尚未推送到远端触发。
+Playwright 需要正在运行的本地 API 和 worker，可用 E2E_PASSWORD 指定测试 admin 密码，默认读取本地凭据文件。首次执行 `npx playwright install chromium`。浏览器 fixture 测试会新建 `e2e-finance.csv` 数据集，不可指向正式生产环境。`.gitea/workflows/ci.yml` 在独立 PostgreSQL 服务容器上执行同样流程（Gitea act_runner 的 job 跑在容器内，服务主机名用服务名 `postgres` 而非 localhost）。`.gitea/workflows/ai-review.yml` 对每个 PR 运行对抗式 AI 审查与自动修复，引擎在 vertai-tech/ai-review-engine（v1 标签），本仓库审查侧重见 `.gitea/ai-review.md`。
 
 原附件只在本机使用，不在仓库中：
 

docs/fix-record-2026-09-26.md

@@ -0,0 +1,30 @@
+# 剩余问题修复记录 · 2026-09-26
+
+对应 `docs/remaining-issues-2026-09-26.html` 第三节（F01–F08）的修复交付记录。
+分支：`fix/remaining-issues-2026-09-26`（基线 `0ac0c4e` 起逐项小步提交）。
+
+## 修复状态
+
+| 编号 | 状态 | 提交 | 说明 |
+| --- | --- | --- | --- |
+| F01 · P0 · 符号链接清理 | 已修复 | `36fd5f3` | 根目录/用户目录/目标文件先 lstat 完整校验（拒绝符号链接与非目录/非普通文件），校验失败不删除任何磁盘文件；失败清理只处理本次登记的受控文件并通过 `O_NOFOLLOW` 目录描述符 unlink，目录被换成符号链接时拒绝删除。测试覆盖用户目录链接、目标链接、运行中目录被换成链接（受保护文件字节不变）、失败后重试成功。 |
+| F02 · P1 · 清理失败与孤儿 | 已修复 | `54dc58d` | 清理受阻时记录不含敏感数据的 WARNING（任务编号+文件名）并保留 `job.path` 使残留可追踪；`cleanup_formula_exports` 新增孤儿 `.xlsx` 回收：超过 24h 保留期且不属于未过期成功任务的文件统一回收，不触碰活动任务。测试含 unlink 权限错误注入→告警→恢复权限后回收且其他任务文件不受影响。 |
+| F03 · P1 · 证据弹窗卡加载 | 已修复 | `6cf236a` | 同期间同页点击直接返回；第二页点击当前期间回第一页刷新；失败显示错误+重试按钮；空结果显示空态行。E2E 覆盖重复点击、快速切换、第二页切期间、失败重试、空列表。 |
+| F04 · P1 · 载入定义丢上下文 | 已修复 | `9a5ab26` | 载入定义时以“用户消息+助手确认”把名称/含义/范围/缺失处理/表达式写入可见对话上下文（需勾选同意才发送），修改后仍需重新确认；选择空选项同步清理定义与上下文。服务端单条消息上限 1000→2000 容纳完整定义。E2E 验证模型请求包含原定义且“乘以 100”修改生效。 |
+| F05 · P1 · 定义库并发与审计 | 已修复 | `0aaa87e` | 迁移 0005 增加 `revision`；保存在单事务内锁定用户所属行：配额（50/数据集）原子化、同名更新必须携带匹配 revision（409 冲突）、审计与变更同事务回滚；保存限流 30/h。前端保存同名定义携带 revision 并明示覆盖行为。PostgreSQL 并发配额用例作为独立门禁（本机 SQLite 自动跳过）。 |
+| F06 · P1 · 工作簿可解释性 | 已修复 | `9962573` | 参数页“产品毛利率”改为只读快照（可编辑列全部为“否”），写明调整位置是计算明细 MarginRate 列；参数页新增维度键-显示名完整对照；使用说明新增输入字典（Revenue/Cost/GrossInput/MarginRate/ListAmount 的单位、来源字段、修改位置、缺失含义）与固定结构边界行。 |
+| F07 · P1 · 对账容差 | 已修复 | `649fec4` | 对账检查页新增容差列：人数严格相等、金额 0.01、比率 0.0001，状态公式按 `ABS(差异)<=容差` 判定；自定义列用 `to_decimal` 白名单表达式解释器（不执行代码）计算 Decimal 服务器参考，不再标注“需业务验收”；参数页写明容差规则与“调整需财务负责人批准”。 |
+| F08 · P1 · 撤权与生命周期 | 已修复 | `5ebe83a` | worker 执行前重新核验账号 `is_active` 与 admin/analyst 角色（撤权/停用后排队任务失败、不发布）；导出状态/下载接口补齐角色白名单（降为 viewer 后 403）；`import_worker` 对超过 15 分钟的 processing 任务重新认领（PG 依赖咨询锁立即恢复），崩溃重启不再无限停留。 |
+
+## 本轮完整测试记录（2026-09-26，SQLite 隔离环境）
+
+- **后端：87 passed，7 skipped，90.3 秒；finance 覆盖率 88.50%（85% 门槛通过）。**
+  跳过项：4 项 LibreOffice 重算（本机未安装 soffice，环境门禁，CI 中执行）、1 项 PostgreSQL 并发配额（依赖 PG 行锁，生产库单独验证）、1 项原附件全量 UI（未设置 `REFERENCE_DATASET_NAME`）。
+- **浏览器 E2E：3 passed，1 skipped，21.0 秒。** 新隔离数据库（`var/e2e-test.sqlite3` + 独立 MEDIA_ROOT），覆盖追问、保存定义、关闭重开、载入定义修改后生成（乘以 100）、下载，含 F03 全部交互回归（重复点击/快速切换/失败重试/空列表/翻页）、桌面与手机。
+- Ruff check / ruff format、TypeScript（tsc --noEmit）、Vite 构建、Django check、`makemigrations --check`（迁移一致）、collectstatic：全部通过。
+- `pip-audit`：无已知漏洞。`npm audit --omit=dev`：本机 registry 访问失败（网络受限，CI 中执行），记录为环境限制项。
+
+## 仍开放的事项（与报告一致，非本轮代码范围）
+
+- 第四节功能缺口（网页主汇总表新列、草稿续聊、原附件 332,830 行导出、高基数与可访问性专项等）与第五节外部门禁（目标 Excel 实机、财务口径签字、AI 真实联调、生产基础设施、PostgreSQL/SLA、安全运营）仍按原报告执行，责任人与关闭证据不变。
+- F07 容差数值（金额 0.01 / 比率 0.0001）为工程默认，仍需财务负责人确认或调整后才视为财务口径关闭。

finance/exporting.py

@@ -21,6 +21,7 @@ from .formula_columns import (
     RESERVED_DATA_HEADERS,
     ensure_available_metrics,
     metrics_used,
+    to_decimal,
     to_excel,
     validate_column,
 )
@@ -30,6 +31,23 @@ FORMULA_VERSION = "3"
 MAX_ROWS = 100_000
 MAX_GROUPS = 200
 
+# 对账比较规则：人数严格相等；金额 0.01；比率 0.0001（一个基点）。
+# 容差属财务口径，调整需财务负责人批准后方可修改。
+RECONCILE_TOLERANCES = {
+    "customers": Decimal("0"),
+    "revenue": Decimal("0.01"),
+    "cost": Decimal("0.01"),
+    "gross": Decimal("0.01"),
+    "self_revenue": Decimal("0.01"),
+    "basket": Decimal("0.0001"),
+    "margin": Decimal("0.0001"),
+    "discount_rate": Decimal("0.0001"),
+    "upgrade_rate": Decimal("0.0001"),
+    "list_basket": Decimal("0.0001"),
+    "roi": Decimal("0.0001"),
+}
+CUSTOM_COLUMN_TOLERANCE = Decimal("0.01")
+
 logger = logging.getLogger(__name__)
 
 
@@ -462,7 +480,12 @@ def build_formula_workbook(ds, member, export_context, output):
         ],
         ["ROI 口径", mapping["roi_formula"], "按导入映射保存的公式分支", "否"],
         ["自营产品集合", _text_cell(parameters, ", ".join(products)), "升单率和自营收入口径", "否"],
-        ["产品毛利率", str(rates), "直接毛利缺失时的可编辑计算输入", "是"],
+        [
+            "产品毛利率（快照）",
+            _text_cell(parameters, str(rates)),
+            "只读快照；导出时已按行写入计算明细 MarginRate 列，调整毛利率请在计算明细页按行修改",
+            "否",
+        ],
         [
             "Excel 数值精度",
             "IEEE-754 双精度",
@@ -471,8 +494,23 @@ def build_formula_workbook(ds, member, export_context, output):
         ],
         ["重算范围", f"最多 {MAX_ROWS:,} 行、{MAX_GROUPS} 组", "超限需缩小导出范围", "否"],
         ["兼容性", "SUMIFS/COUNTIFS", "移除动态数组人数公式；仍需目标 Excel 桌面版本验收", "否"],
+        [
+            "对账容差",
+            "人数严格相等；金额 0.01；比率 0.0001",
+            "对账检查页按指标类型使用独立容差；自定义列按金额容差 0.01。容差调整需财务负责人批准",
+            "否",
+        ],
     ]:
         parameters.append(row)
+    for d in dims:
+        parameters.append(
+            [
+                _text_cell(parameters, f"维度 {d['key']}"),
+                _text_cell(parameters, d["name"]),
+                "维度键与显示名对照（只读）；计算明细与原始证据的列名即维度键",
+                "否",
+            ]
+        )
     for definition in custom_columns:
         parameters.append(
             [
@@ -483,10 +521,33 @@ def build_formula_workbook(ds, member, export_context, output):
             ]
         )
     instructions.append(_styled_row(instructions, ["主题", "说明"], green, white))
+
+    def source_name(role):
+        return next((c["name"] for c in mapping["columns"] if c["role"] == role), "未映射")
+
+    editable_inputs = [
+        ("Revenue", "实际收入", "金额", source_name("revenue"), "空值行不参与求和，汇总按现有行计算"),
+        ("Cost", "成本", "金额", source_name("cost"), "任一行缺失时本组成本及 ROI 等依赖指标不可用"),
+        ("GrossInput", "直接毛利", "金额", source_name("gross"), "空值行改按产品毛利率（lookup）计算毛利"),
+        (
+            "MarginRate",
+            "产品毛利率",
+            "小数比率",
+            "导入确认的产品毛利率表",
+            "仅 GrossSource=lookup 的行参与毛利计算；0.2 表示 20%",
+        ),
+        (
+            "ListAmount",
+            "原价金额",
+            "金额",
+            source_name("list_amount"),
+            "用于扣率与原价客单价；缺失则相关指标不可用",
+        ),
+    ]
     for row in [
         ["用途", "公式分析工作簿；本地修改不回写系统。"],
         ["行粒度", "计算明细每行对应一个 Fact，不等同患者就诊数。"],
-        ["维度列", "列名为数据导入映射中的维度键；显示名称见本页分组维度。"],
+        ["维度列", "列名为数据导入映射中的维度键；键与显示名对照见“参数与口径”页的维度行。"],
         ["毛利", "直接毛利优先，否则收入×产品毛利率；无法计算则不可用。毛利贡献不代表净利润。"],
         ["ROI", "按选定口径；缺成本、成本非正或所需指标缺失时不可用。"],
         ["人数", "客户 token 跨行去重；缺客户编码时不可用。"],
@@ -497,6 +558,25 @@ def build_formula_workbook(ds, member, export_context, output):
         ["变化率", "差额/ABS(基期)；基期为0或期间缺失时不可用。"],
         ["精度", "原始 Decimal 以文本留存；Excel 双精度存在细微误差可能。"],
         ["隐私", "客户标识替换为本文件随机 token，文件仍按敏感财务信息保护。"],
+        [
+            "输入字典",
+            "以下输入可在“计算明细”页直接修改，全部汇总公式与对账实时引用；字段、单位、来源与缺失含义逐项列出。",
+        ],
+        *[
+            [
+                _text_cell(instructions, f"输入 {header}（{label}）"),
+                _text_cell(
+                    instructions,
+                    f"单位：{unit}；来源：{source}；修改位置：计算明细页 {header} 列；缺失含义：{missing}。",
+                ),
+            ]
+            for header, label, unit, source, missing in editable_inputs
+        ],
+        [
+            "固定结构",
+            "Period、维度列、CustomerToken、GrossSource、SelfOperated、CustomerPresent、RowCount 与“人数去重”页均为固定快照；"
+            "修改或增删这些列/行会使汇总、去重与对账失真，需要改变结构时请重新导出。",
+        ],
     ]:
         instructions.append(row)
     for definition in custom_columns:
@@ -600,11 +680,18 @@ def build_formula_workbook(ds, member, export_context, output):
     reconcile.append(
         _styled_row(
             reconcile,
-            ["指标", "分组", "期间侧", "服务器 Decimal 参考", "公式结果", "差异", "状态"],
+            ["指标", "分组", "期间侧", "服务器 Decimal 参考", "公式结果", "差异", "容差", "状态"],
             green,
             white,
         )
     )
+
+    def _reconcile_status(row_number):
+        return (
+            f'=IF(D{row_number}="","无参考",IF(ISNUMBER(F{row_number}),'
+            f'IF(ABS(F{row_number})<=G{row_number},"通过","待复核"),"待复核"))'
+        )
+
     reconcile_row = 2
     for label in labels:
         for side_name, periods, col in (("本期", ctx["current"], "C"), ("基期", ctx["baseline"], "D")):
@@ -621,21 +708,25 @@ def build_formula_workbook(ds, member, export_context, output):
                         "" if value is None else str(value),
                         f"='分析汇总'!{col}{formula_row}",
                         f'=IFERROR(E{reconcile_row}-D{reconcile_row},"")',
-                        f'=IF(D{reconcile_row}="","无参考",IF(ISNUMBER(F{reconcile_row}),IF(ABS(F{reconcile_row})<0.01,"通过","待复核"),"待复核"))',
+                        float(RECONCILE_TOLERANCES[key]),
+                        _reconcile_status(reconcile_row),
                     ]
                 )
                 reconcile_row += 1
             for index, definition in enumerate(custom_columns):
                 formula_row = metric_rows[(label, f"custom_{index}")]
+                # 自定义列用 Decimal 白名单解释器计算服务器参考，不再依赖业务人工验收
+                reference_value = to_decimal(definition["expression"], reference)
                 reconcile.append(
                     [
                         _text_cell(reconcile, definition["name"]),
                         _text_cell(reconcile, label),
                         side_name,
-                        "自定义公式无服务器参考",
+                        "" if reference_value is None else str(reference_value),
                         f"='分析汇总'!{col}{formula_row}",
-                        "",
-                        "需业务验收",
+                        f'=IFERROR(E{reconcile_row}-D{reconcile_row},"")',
+                        float(CUSTOM_COLUMN_TOLERANCE),
+                        _reconcile_status(reconcile_row),
                     ]
                 )
                 reconcile_row += 1
@@ -677,7 +768,12 @@ def process_formula_export(export_id):
         job.save(update_fields=["path", "status", "error", "finished_at", "expires_at"])
         return stats
     except Exception as exc:
-        job.status, job.error, job.finished_at, job.expires_at = "failed", str(exc)[:500], 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:

finance/formula_columns.py

@@ -1,4 +1,5 @@
 import math
+from decimal import Decimal
 
 from .analysis import METRICS
 
@@ -107,3 +108,27 @@ def to_excel(node, references):
     if "number" in node:
         return str(node["number"])
     return f"({to_excel(node['left'], references)}{OPERATORS[node['operator']]}{to_excel(node['right'], references)})"
+
+
+def to_decimal(node, values):
+    """按已校验的白名单表达式树用 Decimal 计算服务器参考值。
+
+    只遍历 metric/number/四则运算节点，不执行任何代码；依赖指标缺失或除零时
+    返回 None，与 Excel 侧 #N/A 的语义一致。
+    """
+    if "metric" in node:
+        return values[node["metric"]]
+    if "number" in node:
+        return Decimal(str(node["number"]))
+    left, right = to_decimal(node["left"], values), to_decimal(node["right"], values)
+    if left is None or right is None:
+        return None
+    if node["operator"] == "add":
+        return left + right
+    if node["operator"] == "subtract":
+        return left - right
+    if node["operator"] == "multiply":
+        return left * right
+    if right == 0:
+        return None
+    return left / right

finance/formula_views.py

@@ -3,14 +3,17 @@ from pathlib import Path
 
 import httpx
 from django.conf import settings
+from django.db import IntegrityError, transaction
 from django.http import FileResponse, JsonResponse
 from django.utils import timezone
 
 from .analysis import METRICS, context, dimensions
 from .formula_columns import available_metrics, ensure_available_metrics, validate_column
-from .models import FormulaDefinition, FormulaExport
+from .models import FormulaDefinition, FormulaExport, Membership
 from .views import APIError, audit, body, dataset, endpoint, limit
 
+MAX_DEFINITIONS_PER_DATASET = 50
+
 
 def design_column(messages, ds, ctx):
     if not (settings.AI_ENABLED and settings.OPENAI_API_KEY and settings.OPENAI_MODEL):
@@ -198,6 +201,7 @@ def formula_definitions(request, dataset_id):
                     {
                         "id": str(item.pk),
                         "definition": item.definition,
+                        "revision": item.revision,
                         "updated_at": item.updated_at.isoformat(),
                     }
                     for item in items
@@ -205,28 +209,54 @@ def formula_definitions(request, dataset_id):
             }
         )
 
-    definition = validate_column(body(request).get("definition"))
+    data = body(request)
+    definition = validate_column(data.get("definition"))
     try:
         ensure_available_metrics(ds, definition)
     except ValueError as exc:
         raise APIError(str(exc)) from exc
     normalized_name = definition["name"].strip().casefold()
-    existing = FormulaDefinition.objects.filter(
-        dataset=ds, user=request.user, normalized_name=normalized_name
-    ).first()
-    if not existing and FormulaDefinition.objects.filter(dataset=ds, user=request.user).count() >= 50:
-        raise APIError("每个数据集最多保存 50 个公式定义")
-    item, created = FormulaDefinition.objects.update_or_create(
-        dataset=ds,
-        user=request.user,
-        normalized_name=normalized_name,
-        defaults={"definition": definition},
-    )
-    audit(request, "formula_definition.save", item.pk, {"dataset": str(ds.pk), "name": definition["name"]})
+    limit(f"formula-definition:{request.user.pk}", 30, 3600)
+    with transaction.atomic():
+        # 锁定稳定的用户所属行序列化同一用户的并发保存，使配额检查与写入原子化。
+        Membership.objects.select_for_update().get(pk=request.member.pk)
+        existing = (
+            FormulaDefinition.objects.select_for_update()
+            .filter(dataset=ds, user=request.user, normalized_name=normalized_name)
+            .first()
+        )
+        if existing:
+            # 同名更新按 revision 做乐观并发控制，避免两个窗口相互静默覆盖。
+            if data.get("revision") != existing.revision:
+                raise APIError("定义已在其他窗口更新，请刷新后重试", 409)
+            existing.definition, existing.revision = definition, existing.revision + 1
+            existing.save(update_fields=["definition", "revision", "updated_at"])
+            item, created = existing, False
+        else:
+            if (
+                FormulaDefinition.objects.filter(dataset=ds, user=request.user).count()
+                >= MAX_DEFINITIONS_PER_DATASET
+            ):
+                raise APIError(f"每个数据集最多保存 {MAX_DEFINITIONS_PER_DATASET} 个公式定义")
+            try:
+                item = FormulaDefinition.objects.create(
+                    dataset=ds,
+                    user=request.user,
+                    normalized_name=normalized_name,
+                    definition=definition,
+                )
+            except IntegrityError as exc:
+                raise APIError("定义保存发生并发冲突，请重试", 409) from exc
+            created = True
+        # 审计与定义变更在同一事务内提交；审计失败时变更一并回滚。
+        audit(
+            request, "formula_definition.save", item.pk, {"dataset": str(ds.pk), "name": definition["name"]}
+        )
     return JsonResponse(
         {
             "id": str(item.pk),
             "definition": item.definition,
+            "revision": item.revision,
             "updated_at": item.updated_at.isoformat(),
         },
         status=201 if created else 200,
@@ -242,8 +272,9 @@ def formula_definition_delete(request, dataset_id, definition_id):
     if not item:
         raise APIError("公式定义不存在", 404)
     name = item.definition.get("name", "")
-    audit(request, "formula_definition.delete", item.pk, {"dataset": str(ds.pk), "name": name})
-    item.delete()
+    with transaction.atomic():
+        item.delete()
+        audit(request, "formula_definition.delete", item.pk, {"dataset": str(ds.pk), "name": name})
     return JsonResponse({"deleted": True})
 
 

finance/management/commands/cleanup_formula_exports.py

@@ -23,9 +23,7 @@ class Command(BaseCommand):
     def handle(self, *args, **options):
         while True:
             jobs, files, orphans = self.clean_expired()
-            self.stdout.write(
-                f"Expired {jobs} jobs; removed {files} files; reclaimed {orphans} orphan files"
-            )
+            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"]))

finance/migrations/0005_formuladefinition_revision.py

@@ -0,0 +1,17 @@
+# Generated by Django 5.2.17 on 2026-09-25 17:30
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("finance", "0004_formuladefinition"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="formuladefinition",
+            name="revision",
+            field=models.PositiveIntegerField(default=1),
+        ),
+    ]

finance/models.py

@@ -121,6 +121,7 @@ class FormulaDefinition(models.Model):
     user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
     normalized_name = models.CharField(max_length=320)
     definition = models.JSONField()
+    revision = models.PositiveIntegerField(default=1)
     created_at = models.DateTimeField(auto_now_add=True)
     updated_at = models.DateTimeField(auto_now=True)
 

frontend/src/dialogs.tsx

@@ -31,6 +31,7 @@ type FormulaExpression =
 type SavedFormulaDefinition = {
   id: string;
   definition: Record<string, unknown>;
+  revision: number;
   updated_at: string;
 };
 
@@ -802,10 +803,18 @@ export function FormulaExportDialog({
     setBusy(true);
     setError("");
     try {
+      // 同名保存按服务端匹配规则覆盖旧版本；带上已知版本的 revision，
+      // 其他窗口先更新过则服务端返回 409，避免静默覆盖。
+      const normalizedName = (value: string) => value.trim().toLowerCase();
+      const existing = definitions.find(
+        (item) =>
+          normalizedName(String(item.definition.name)) ===
+          normalizedName(String(column.name)),
+      );
       const saved = await api<SavedFormulaDefinition>(
         `/datasets/${ds.id}/formula-definitions`,
         "POST",
-        { definition: column },
+        { definition: column, revision: existing?.revision },
       );
       setDefinitions((items) =>
         [...items.filter((item) => item.id !== saved.id), saved].sort((a, b) =>
@@ -914,6 +923,11 @@ export function FormulaExportDialog({
             </button>
           </div>
         )}
+        {definitions.length > 0 && !task && (
+          <p className="muted">
+            保存同名定义会覆盖原版本；若已在其他窗口被修改，保存将提示刷新后重试。
+          </p>
+        )}
         <label className="check">
           <input
             type="checkbox"

scripts/dev_up.sh

@@ -0,0 +1,124 @@
+#!/usr/bin/env bash
+# 一键启动本地开发环境：迁移 → 初始化管理员 → 构建前端 → 启动 API+worker → 打开页面。
+# 默认使用 SQLite 轻量开发库（var/db.sqlite3）；设置 DATABASE_URL 则使用 PostgreSQL。
+# Ctrl-C 或 --stop 停止本脚本启动的进程。
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "$ROOT"
+
+PORT="${PORT:-8000}"
+SERVER_PID_FILE="var/dev-server.pid"
+WORKER_PID_FILE="var/dev-worker.pid"
+SERVER_LOG="var/dev-server.log"
+WORKER_LOG="var/dev-worker.log"
+
+die() { echo "错误：$*" >&2; exit 1; }
+
+detect_python() {
+  if [ -x ".venv/bin/python" ]; then
+    PY=".venv/bin/python"
+  elif command -v uv >/dev/null 2>&1; then
+    PY="uv run python"
+  else
+    die "未找到 .venv/bin/python 或 uv，请先安装依赖（uv sync）"
+  fi
+}
+
+read_pidfile() { [ -f "$1" ] && cat "$1" || true; }
+
+stop_all() {
+  local stopped=0 pid
+  for file in "$SERVER_PID_FILE" "$WORKER_PID_FILE"; do
+    pid="$(read_pidfile "$file")"
+    if [ -n "$pid" ] && kill -0 "$pid" 2>/dev/null; then
+      kill "$pid" 2>/dev/null || true
+      stopped=1
+    fi
+    rm -f "$file"
+  done
+  [ "$stopped" = 1 ] && echo "已停止本地开发服务" || echo "没有正在运行的开发服务"
+}
+
+health_ok() { curl -fsS --max-time 2 "http://127.0.0.1:$PORT/api/health" >/dev/null 2>&1; }
+
+open_page() {
+  local url="http://127.0.0.1:$PORT/"
+  if command -v open >/dev/null 2>&1; then open "$url"
+  elif command -v xdg-open >/dev/null 2>&1; then xdg-open "$url" >/dev/null 2>&1 || true
+  fi
+}
+
+case "${1:-}" in
+  --stop)
+    stop_all
+    exit 0
+    ;;
+  "" | --no-open) ;;
+  *) die "未知参数：$1（支持 --stop / --no-open）" ;;
+esac
+
+detect_python
+mkdir -p var
+
+if health_ok; then
+  echo "已在 $PORT 端口运行，直接打开页面"
+  [ "${1:-}" != "--no-open" ] && open_page
+  exit 0
+fi
+
+if [ -z "${DATABASE_URL:-}" ]; then
+  echo "未设置 DATABASE_URL，使用 SQLite 轻量开发库 var/db.sqlite3（仅限页面测试与开发回归）"
+fi
+
+echo "执行数据库迁移…"
+$PY manage.py migrate --no-input >/dev/null
+
+if [ ! -f var/local-access.txt ]; then
+  echo "初始化本地管理员…"
+  PYTHONPATH="$ROOT" $PY scripts/init_local.py
+fi
+
+if [ ! -d frontend/dist ]; then
+  echo "构建前端（首次）…"
+  npm run build >/dev/null
+fi
+
+echo "启动 worker（日志 ${WORKER_LOG}）…"
+nohup $PY manage.py import_worker >"$WORKER_LOG" 2>&1 &
+echo $! >"$WORKER_PID_FILE"
+
+echo "启动 API 服务…"
+nohup $PY manage.py runserver "127.0.0.1:$PORT" --noreload >"$SERVER_LOG" 2>&1 &
+echo $! >"$SERVER_PID_FILE"
+
+cleanup() {
+  stop_all >/dev/null 2>&1 || true
+}
+trap cleanup INT TERM
+
+for _ in $(seq 1 30); do
+  if health_ok; then
+    break
+  fi
+  if [ -n "$(read_pidfile "$SERVER_PID_FILE")" ] && ! kill -0 "$(read_pidfile "$SERVER_PID_FILE")" 2>/dev/null; then
+    tail -20 "$SERVER_LOG" >&2
+    die "API 服务启动失败（完整日志：${SERVER_LOG}）"
+  fi
+  sleep 1
+done
+health_ok || { tail -20 "$SERVER_LOG" >&2; die "健康检查超时（日志：${SERVER_LOG}）"; }
+
+echo
+echo "本地开发环境已就绪：http://127.0.0.1:$PORT"
+[ -f var/local-access.txt ] && echo "登录凭据：var/local-access.txt（用户名 admin，权限 0600）"
+if [ -z "${DATABASE_URL:-}" ]; then
+  echo "提示：当前为 SQLite 轻量模式；高精度金额聚合与生产验收请按 README 配置 PostgreSQL。"
+fi
+echo "日志：${SERVER_LOG} / ${WORKER_LOG} · Ctrl-C 或 scripts/dev_up.sh --stop 停止"
+echo
+
+[ "${1:-}" != "--no-open" ] && open_page
+
+wait "$(read_pidfile "$SERVER_PID_FILE")" || true
+cleanup

tests/test_assistant_api.py

@@ -168,16 +168,21 @@ def test_formula_definition_library_is_private_validated_and_audited(workspace):
     saved = client.post(url, data={"definition": definition}, content_type="application/json")
     assert saved.status_code == 201
     definition_id = saved.json()["id"]
+    assert saved.json()["revision"] == 1
     assert client.get(url).json()["items"][0]["definition"] == definition
 
     updated = client.post(
         url,
-        data={"definition": {**definition, "name": "  收入成本差  ", "meaning": "更新说明"}},
+        data={
+            "definition": {**definition, "name": "  收入成本差  ", "meaning": "更新说明"},
+            "revision": 1,
+        },
         content_type="application/json",
     )
     assert updated.status_code == 200
     assert FormulaDefinition.objects.filter(dataset=ds, user=ds.created_by).count() == 1
     assert updated.json()["definition"]["meaning"] == "更新说明"
+    assert updated.json()["revision"] == 2
 
     other = Client()
     other_user = ds.created_by.__class__.objects.create_user(

tests/test_formula_columns.py

@@ -97,8 +97,11 @@ def test_formula_workbook_exposes_custom_metric_as_recalculable_column(formula_d
     custom = workbook["自定义指标"]
     assert custom["B1"].value == "收入成本差（本期）"
     assert custom["B2"].value == f"='分析汇总'!C{metric_row}"
-    assert "含义：实际收入减成本" in workbook["使用说明"]["B12"].value
-    assert "范围：按所选分组及本期、基期计算" in workbook["使用说明"]["B12"].value
+    custom_note = next(
+        str(row[1]) for row in workbook["使用说明"].iter_rows(values_only=True) if row[0] == "收入成本差"
+    )
+    assert "含义：实际收入减成本" in custom_note
+    assert "范围：按所选分组及本期、基期计算" in custom_note
 
 
 def test_custom_expression_rejects_formula_text_and_unknown_metrics():

tests/test_formula_definitions_concurrency.py

@@ -0,0 +1,123 @@
+from unittest.mock import patch
+
+import pytest
+from django.db import connection
+
+from finance.models import FormulaDefinition
+
+pytestmark = pytest.mark.django_db
+
+
+def column_definition(name="收入成本差"):
+    return {
+        "name": name,
+        "meaning": "实际收入减成本",
+        "scope": "所选分组与期间",
+        "missing": "结果不可用（#N/A）；零分母不可用",
+        "expression": {
+            "operator": "subtract",
+            "left": {"metric": "revenue"},
+            "right": {"metric": "cost"},
+        },
+    }
+
+
+def _post(client, url, definition, revision=None):
+    return client.post(
+        url,
+        data={"definition": definition, **({"revision": revision} if revision is not None else {})},
+        content_type="application/json",
+    )
+
+
+def test_same_name_update_requires_matching_revision(workspace):
+    ds, _, client, *_ = workspace
+    url = f"/api/datasets/{ds.pk}/formula-definitions"
+
+    saved = _post(client, url, column_definition())
+    assert saved.status_code == 201
+    item_id, revision = saved.json()["id"], saved.json()["revision"]
+
+    # 缺失或过期的 revision 都不能静默覆盖其他窗口的修改
+    assert _post(client, url, column_definition()).status_code == 409
+    assert _post(client, url, column_definition(), revision=999).status_code == 409
+
+    updated = _post(client, url, column_definition(), revision=revision)
+    assert updated.status_code == 200
+    assert updated.json()["revision"] == revision + 1
+
+    # 基于旧版本的再次更新必须被拒绝，两个标签页不互相静默丢失
+    stale = _post(client, url, column_definition(), revision=revision)
+    assert stale.status_code == 409
+    current = FormulaDefinition.objects.get(pk=item_id)
+    assert current.revision == revision + 1
+    # 列表接口暴露当前 revision 供前端携带
+    listed = client.get(url).json()["items"]
+    assert listed[0]["revision"] == revision + 1
+
+
+def test_definition_quota_is_enforced_and_release_allows_new_save(workspace):
+    ds, _, client, *_ = workspace
+    url = f"/api/datasets/{ds.pk}/formula-definitions"
+    for index in range(50):
+        FormulaDefinition.objects.create(
+            dataset=ds,
+            user=ds.created_by,
+            normalized_name=f"定义{index:02d}",
+            definition=column_definition(f"定义{index:02d}"),
+        )
+
+    rejected = _post(client, url, column_definition("第 51 个不同名定义"))
+    assert rejected.status_code == 400
+    assert "50" in rejected.json()["error"]
+    assert FormulaDefinition.objects.filter(dataset=ds).count() == 50
+
+    # 同名更新不受配额限制；删除后名额释放可再保存新名称
+    assert _post(client, url, column_definition("定义00"), revision=1).status_code == 200
+    FormulaDefinition.objects.filter(normalized_name="定义00").delete()
+    assert _post(client, url, column_definition("第 51 个不同名定义")).status_code == 201
+
+
+def test_audit_failure_rolls_back_definition_change(workspace):
+    ds, _, client, *_ = workspace
+    url = f"/api/datasets/{ds.pk}/formula-definitions"
+
+    with patch("finance.formula_views.audit", side_effect=RuntimeError("audit unavailable")):
+        with pytest.raises(RuntimeError, match="audit unavailable"):
+            _post(client, url, column_definition())
+
+    assert FormulaDefinition.objects.filter(dataset=ds).count() == 0
+
+
+def test_concurrent_distinct_saves_cannot_exceed_quota_on_postgres(workspace):
+    if connection.vendor != "postgresql":
+        pytest.skip("并发配额依赖 PostgreSQL 行锁语义，生产库单独验证")
+    import threading
+
+    from django.test import Client
+
+    ds, member, *_ = workspace
+    user = ds.created_by
+    for index in range(49):
+        FormulaDefinition.objects.create(
+            dataset=ds,
+            user=user,
+            normalized_name=f"定义{index:02d}",
+            definition=column_definition(f"定义{index:02d}"),
+        )
+
+    barrier = threading.Barrier(4)
+
+    def save(index):
+        client = Client()
+        assert client.login(username=user.username, password="Finance-TestPassword991!")
+        barrier.wait()
+        _post(client, f"/api/datasets/{ds.pk}/formula-definitions", column_definition(f"并发{index}"))
+
+    threads = [threading.Thread(target=save, args=(index,)) for index in range(4)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+
+    assert FormulaDefinition.objects.filter(dataset=ds).count() == 50

tests/test_formula_reconcile_guard.py

@@ -20,11 +20,12 @@ def test_reconcile_status_guards_non_numeric_difference(workspace):
         output,
     )
     workbook = load_workbook(BytesIO(output.getvalue()), data_only=False)
+    status_cells = [row[7] for row in workbook["对账检查"].iter_rows(min_row=2)]
     formulas = [
-        row[6].value
-        for row in workbook["对账检查"].iter_rows(min_row=2)
-        if isinstance(row[6].value, str) and row[6].value.startswith("=IF(D")
+        cell.value for cell in status_cells if isinstance(cell.value, str) and cell.value.startswith("=IF(D")
     ]
     assert formulas
     assert all("ISNUMBER(F" in formula for formula in formulas)
     assert all('"通过"' in formula for formula in formulas)
+    # 容差列独立成列，状态按容差比较而不是统一 0.01
+    assert all(row[6].value is not None for row in workbook["对账检查"].iter_rows(min_row=2))

tests/test_formula_reconcile_tolerances.py

@@ -0,0 +1,236 @@
+import os
+import shutil
+import subprocess
+from decimal import Decimal
+from io import BytesIO
+
+import pytest
+from openpyxl import load_workbook
+
+from finance.exporting import CUSTOM_COLUMN_TOLERANCE, RECONCILE_TOLERANCES, build_formula_workbook
+from finance.formula_columns import to_decimal
+
+pytestmark = pytest.mark.django_db
+
+
+def custom_column(name="收入成本差"):
+    return {
+        "name": name,
+        "meaning": "实际收入减成本",
+        "scope": "按所选分组及本期、基期计算",
+        "missing": "结果不可用（#N/A）；零分母不可用",
+        "expression": {
+            "operator": "subtract",
+            "left": {"metric": "revenue"},
+            "right": {"metric": "cost"},
+        },
+    }
+
+
+def test_to_decimal_evaluator_golden_samples():
+    values = {
+        "revenue": Decimal("999999999999.99"),
+        "cost": Decimal("0.0000001"),
+        "customers": Decimal("7"),
+    }
+    # 大数与多层表达式：((收入 - 收入) + 成本×2) ÷ (成本×2) = 1，抵消后不丢失精度
+    cancellation = {
+        "operator": "divide",
+        "left": {
+            "operator": "add",
+            "left": {
+                "operator": "subtract",
+                "left": {"metric": "revenue"},
+                "right": {"metric": "revenue"},
+            },
+            "right": {"operator": "multiply", "left": {"metric": "cost"}, "right": {"number": 2}},
+        },
+        "right": {"operator": "multiply", "left": {"metric": "cost"}, "right": {"number": 2}},
+    }
+    assert to_decimal(cancellation, values) == Decimal("1")
+    # 极小值与比率
+    assert to_decimal({"metric": "cost"}, values) == Decimal("0.0000001")
+    assert to_decimal(
+        {"operator": "divide", "left": {"metric": "revenue"}, "right": {"metric": "customers"}}, values
+    ) == Decimal("999999999999.99") / Decimal("7")
+    # 除零与缺失指标返回 None（对应 Excel #N/A）
+    assert (
+        to_decimal({"operator": "divide", "left": {"metric": "revenue"}, "right": {"number": 0}}, values)
+        is None
+    )
+    assert (
+        to_decimal(
+            {"operator": "add", "left": {"metric": "cost"}, "right": {"metric": "customers"}},
+            {**values, "customers": None},
+        )
+        is None
+    )
+    # 负数与常数组合
+    assert to_decimal(
+        {"operator": "subtract", "left": {"number": -2.5}, "right": {"metric": "cost"}}, values
+    ) == Decimal("-2.5") - Decimal("0.0000001")
+
+
+def test_reconcile_uses_independent_tolerances_per_metric_kind(workspace):
+    ds, member, *_ = workspace
+    output = BytesIO()
+    build_formula_workbook(
+        ds,
+        member,
+        {"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
+        output,
+    )
+    workbook = load_workbook(BytesIO(output.getvalue()), data_only=False)
+    sheet = workbook["对账检查"]
+    rows = list(sheet.iter_rows(values_only=True))
+    assert list(rows[0]) == [
+        "指标",
+        "分组",
+        "期间侧",
+        "服务器 Decimal 参考",
+        "公式结果",
+        "差异",
+        "容差",
+        "状态",
+    ]
+
+    def find(metric, side="本期"):
+        return next(row for row in rows[1:] if row[0] == metric and row[2] == side)
+
+    assert find("消费人数")[6] == 0  # 人数严格相等
+    assert find("实际收入")[6] == pytest.approx(0.01)
+    assert find("毛利贡献")[6] == pytest.approx(0.01)
+    assert find("毛利率")[6] == pytest.approx(0.0001)
+    assert find("ROI")[6] == pytest.approx(0.0001)
+    assert find("客单价")[6] == pytest.approx(0.0001)
+    # 状态公式按容差列比较（<=G），不再统一 <0.01
+    status = next(
+        cell.value for cell in sheet[2] if isinstance(cell.value, str) and cell.value.startswith("=IF(D")
+    )
+    assert "<=G2" in status
+    # 参数页写明对账规则，容差调整需财务批准
+    parameters = {str(row[0]): row for row in workbook["参数与口径"].iter_rows(values_only=True) if row[0]}
+    assert "人数严格相等" in parameters["对账容差"][1]
+    assert parameters["对账容差"][3] == "否"
+
+
+def test_custom_columns_reconcile_against_decimal_reference(workspace):
+    ds, member, *_ = workspace
+    output = BytesIO()
+    build_formula_workbook(
+        ds,
+        member,
+        {
+            "start": "2026-08",
+            "end": "2026-08",
+            "comparison": "yoy",
+            "group": "department",
+            "custom_columns": [custom_column()],
+        },
+        output,
+    )
+    workbook = load_workbook(BytesIO(output.getvalue()), data_only=False)
+    rows = list(workbook["对账检查"].iter_rows(values_only=True))
+    custom_rows = [row for row in rows[1:] if row[0] == "收入成本差"]
+    assert custom_rows
+    pediatrics = next(row for row in custom_rows if row[1] == "儿科" and row[2] == "本期")
+    assert Decimal(pediatrics[3]) == Decimal("60")  # 100 - 40
+    assert pediatrics[6] == pytest.approx(float(CUSTOM_COLUMN_TOLERANCE))
+    status = next(
+        row[7]
+        for row in workbook["对账检查"].iter_rows(min_row=2, values_only=True)
+        if row[0] == "收入成本差" and row[1] == "儿科" and row[2] == "本期"
+    )
+    assert "<=G" in status and "通过" in status
+    # 旧的占位文案不再出现
+    assert all("自定义公式无服务器参考" not in str(row[3]) for row in custom_rows)
+    assert all(row[7] != "需业务验收" for row in custom_rows)
+
+
+def test_custom_column_reference_missing_when_dependency_unavailable(workspace):
+    from finance.models import Fact
+
+    ds, member, *_ = workspace
+    # 外科 2026-08 成本置空 → cost 不可用 → 收入成本差参考为 None → 无参考
+    Fact.objects.filter(dataset=ds, period="2026-08", dimensions__department="外科").update(cost=None)
+    output = BytesIO()
+    build_formula_workbook(
+        ds,
+        member,
+        {
+            "start": "2026-08",
+            "end": "2026-08",
+            "comparison": "yoy",
+            "group": "department",
+            "custom_columns": [custom_column()],
+        },
+        output,
+    )
+    workbook = load_workbook(BytesIO(output.getvalue()), data_only=False)
+    surgery = next(
+        row
+        for row in workbook["对账检查"].iter_rows(min_row=2, values_only=True)
+        if row[0] == "收入成本差" and row[1] == "外科" and row[2] == "本期"
+    )
+    assert surgery[3] in ("", None)  # 无服务器参考
+    assert "无参考" in surgery[7]
+
+
+def test_reconcile_tolerances_cover_every_metric():
+    from finance.analysis import METRICS
+
+    assert set(RECONCILE_TOLERANCES) == set(METRICS)
+    assert RECONCILE_TOLERANCES["customers"] == 0
+    assert all(
+        tolerance in (Decimal("0.01"), Decimal("0.0001"))
+        for key, tolerance in RECONCILE_TOLERANCES.items()
+        if key != "customers"
+    )
+
+
+def test_reconcile_passes_in_libreoffice_with_per_metric_tolerances(workspace, tmp_path):
+    executable = os.environ.get("SOFFICE_BIN") or shutil.which("soffice")
+    if not executable:
+        pytest.skip("set SOFFICE_BIN to run the LibreOffice recalculation check")
+    ds, member, *_ = workspace
+    source, output = tmp_path / "source", tmp_path / "out"
+    source.mkdir()
+    output.mkdir()
+    workbook_path = source / "reconcile.xlsx"
+    with workbook_path.open("wb") as file:
+        build_formula_workbook(
+            ds,
+            member,
+            {
+                "start": "2026-08",
+                "end": "2026-08",
+                "comparison": "yoy",
+                "group": "department",
+                "custom_columns": [custom_column()],
+            },
+            file,
+        )
+    profile = tmp_path / "profile"
+    result = subprocess.run(
+        [
+            executable,
+            "--headless",
+            f"-env:UserInstallation={profile.as_uri()}",
+            "--convert-to",
+            "xlsx",
+            "--outdir",
+            str(output),
+            str(workbook_path),
+        ],
+        capture_output=True,
+        text=True,
+        timeout=120,
+        check=False,
+    )
+    assert result.returncode == 0, result.stderr or result.stdout
+    values = load_workbook(output / workbook_path.name, data_only=True)
+    statuses = [
+        (row[0], row[1], row[2], row[7]) for row in values["对账检查"].iter_rows(min_row=2, values_only=True)
+    ]
+    assert statuses
+    assert all(status in ("通过", "无参考") for *_head, status in statuses), statuses

tests/test_formula_workbook_docs.py

@@ -0,0 +1,155 @@
+from io import BytesIO
+
+import pytest
+from openpyxl import load_workbook
+
+from finance.exporting import build_formula_workbook
+
+pytestmark = pytest.mark.django_db
+
+EDITABLE_INPUT_HEADERS = ["Revenue", "Cost", "GrossInput", "MarginRate", "ListAmount"]
+
+
+def _workbook(workspace, **extra):
+    ds, member, *_ = workspace
+    output = BytesIO()
+    build_formula_workbook(
+        ds,
+        member,
+        {"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department", **extra},
+        output,
+    )
+    return load_workbook(BytesIO(output.getvalue()), data_only=False)
+
+
+def test_parameter_sheet_is_read_only_snapshot(workspace):
+    workbook = _workbook(workspace)
+    rows = list(workbook["参数与口径"].iter_rows(values_only=True))
+    header = list(rows[0])
+    editable = header.index("可编辑")
+    assert all(row[editable] == "否" for row in rows[1:] if row[0])
+
+    margin = next(row for row in rows if row[0] and "产品毛利率" in str(row[0]))
+    assert "只读" in margin[2]
+    assert "MarginRate" in margin[2]
+    assert "计算明细" in margin[2]
+
+    # 维度键与显示名完整对照
+    texts = [str(row[0]) for row in rows if row[0]]
+    assert "维度 department" in texts and "维度 product" in texts
+    department = next(row for row in rows if row[0] == "维度 department")
+    assert department[1] == "科室"
+    product = next(row for row in rows if row[0] == "维度 product")
+    assert product[1] == "产品一级分类"
+
+
+def test_instructions_document_editable_inputs_with_real_dependencies(workspace):
+    workbook = _workbook(workspace)
+    instructions = {
+        str(row[0]): str(row[1]) for row in workbook["使用说明"].iter_rows(values_only=True) if row[0]
+    }
+
+    data_headers = [cell.value for cell in workbook["计算明细"][1]]
+    sources = {
+        "Revenue": "实际收入",
+        "Cost": "成本",
+        "GrossInput": "毛利贡献",
+        "MarginRate": "产品毛利率表",
+        "ListAmount": "应收金额",
+    }
+    for header in EDITABLE_INPUT_HEADERS:
+        key = next(k for k in instructions if k.startswith(f"输入 {header}（"))
+        text = instructions[key]
+        assert f"计算明细页 {header} 列" in text
+        assert sources[header] in text
+        assert header in data_headers
+
+    # 每个可修改输入都被汇总公式真实引用
+    summary_formula_cells = [
+        cell.value
+        for row in workbook["分析汇总"].iter_rows(min_row=6)
+        for cell in row
+        if isinstance(cell.value, str) and cell.value.startswith("=IF")
+    ]
+    joined = "\n".join(summary_formula_cells)
+    for header in ("Revenue", "Cost", "ListAmount"):
+        assert f"Data[{header}]" in joined
+    detail_formula = next(
+        cell.value
+        for row in workbook["计算明细"].iter_rows(min_row=2)
+        for cell in row
+        if isinstance(cell.value, str) and cell.value.startswith("=IF(")
+    )
+    assert "GrossInput" not in detail_formula  # 列名按相对引用参与 GrossCalc
+    assert detail_formula.startswith("=IF(")
+
+    boundary = instructions["固定结构"]
+    assert "人数去重" in boundary and "固定快照" in boundary
+    assert "重新导出" in boundary
+    assert "参数与口径" in instructions["维度列"]
+
+
+def test_margin_rate_lookup_edit_changes_gross_via_detail_recalculation(workspace, tmp_path):
+    import os
+    import shutil
+
+    from finance.models import Fact
+
+    if not (os.environ.get("SOFFICE_BIN") or shutil.which("soffice")):
+        pytest.skip("set SOFFICE_BIN to run the LibreOffice recalculation check")
+
+    ds, member, *_ = workspace
+    # 去掉直接毛利，使毛利全部走 lookup 毛利率
+    Fact.objects.filter(dataset=ds).update(gross=None)
+    mapping = {**ds.mapping, "margin_rates": {"医疗服务": "0.3"}}
+    ds.mapping = mapping
+    ds.save(update_fields=["mapping"])
+
+    source, output = tmp_path / "source", tmp_path / "out"
+    source.mkdir()
+    output.mkdir()
+    workbook_path = source / "lookup.xlsx"
+    with workbook_path.open("wb") as file:
+        build_formula_workbook(
+            ds,
+            member,
+            {"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
+            file,
+        )
+
+    editable = load_workbook(workbook_path)
+    # MarginRate 列在维度列之后：G 列（期间+2维度=3 列后第 4 个保留列）
+    header_row = [cell.value for cell in editable["计算明细"][1]]
+    rate_column = header_row.index("MarginRate") + 1
+    editable["计算明细"].cell(row=2, column=rate_column).value = 0.5
+    editable.save(workbook_path)
+
+    import subprocess
+
+    profile = tmp_path / "profile"
+    result = subprocess.run(
+        [
+            os.environ.get("SOFFICE_BIN") or shutil.which("soffice"),
+            "--headless",
+            f"-env:UserInstallation={profile.as_uri()}",
+            "--convert-to",
+            "xlsx",
+            "--outdir",
+            str(output),
+            str(workbook_path),
+        ],
+        capture_output=True,
+        text=True,
+        timeout=120,
+        check=False,
+    )
+    assert result.returncode == 0, result.stderr or result.stdout
+    values = load_workbook(output / workbook_path.name, data_only=True)
+    summary = values["分析汇总"]
+    gross_row = next(
+        row
+        for row in range(6, summary.max_row + 1)
+        if summary.cell(row, 1).value == "毛利贡献" and summary.cell(row, 2).value == "儿科"
+    )
+    # 2026-08 儿科收入 100，lookup 毛利率改为 0.5 后毛利应为 50
+    assert summary.cell(gross_row, 3).value == pytest.approx(50)

tests/test_worker_paths.py

@@ -6,7 +6,7 @@ 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
 
@@ -20,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(
@@ -88,26 +92,21 @@ def test_status_and_download_deny_revoked_viewer(workspace, settings):
     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
-    )
+    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
-    )
+    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
+        assert client.get(f"/api/datasets/{ds.pk}/formula-exports/{job.pk}/download").status_code == code, (
+            status
+        )
     succeeded = _export_job(
         ds,
         status="succeeded",
@@ -115,18 +114,13 @@ def test_half_finished_export_cannot_be_downloaded(workspace, settings):
         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
-    )
+    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)
-    )
+    FormulaExport.objects.filter(pk=job.pk).update(created_at=timezone.now() - timedelta(minutes=20))
 
     call_command("import_worker", once=True)