HEL-282: 写库路径事务提交与审计留痕自查修复

同类未提交即 close 回滚、业务与审计拆成两笔事务的路径一并收进可嵌套事务边界。

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: multica-agent <github@multica.ai>

.gitignore

@@ -28,3 +28,9 @@ package-lock.json
 # local vendor for agent test env (not shipped)
 .vendor/
 vendor_wheels/
+
+# 正式环境部署产物（证书私钥、备份）
+deploy/tls/certs/
+backups/
+ALERT.log
+backup.log

deploy/.env.example

@@ -1,6 +1,10 @@
 # 测试环境变量示例。复制为 deploy/.env 后填写。
 # 正式环境口令不得提交进仓库。
 
+# 容器运行 uid/gid：必须与 data/ 目录属主一致（宿主机 `id <用户>` 查看）
+APP_UID=10001
+APP_GID=10001
+
 APP_HOST=0.0.0.0
 APP_PORT=4173
 APP_DB_PATH=/app/data/app.db
@@ -12,3 +16,16 @@ APP_BOOTSTRAP_ADMIN_PASSWORD=change-me-in-local-env
 
 # 内网测试可暂时关闭登录失败锁定；正式环境不得开启
 # APP_LOGIN_RATE_LIMIT_DISABLED=1
+
+# ---- 同宿主机第二套环境（正式）必须设置的隔离参数 ----
+# compose 项目名（默认 deploy；正式环境改为 caiwuzongzhang-prod）
+# COMPOSE_PROJECT_NAME=caiwuzongzhang-prod
+# 应用容器名与宿主机端口映射（正式默认只绑本机回环，由反代对外）
+# APP_CONTAINER_NAME=caiwuzongzhang-prod-app
+# APP_PORT_MAP=127.0.0.1:4174:4173
+
+# ---- 正式 HTTPS 反代（COMPOSE_PROFILES=tls 启用 proxy 服务）----
+# COMPOSE_PROFILES=tls
+# PROXY_CONTAINER_NAME=caiwuzongzhang-proxy
+# PROXY_PORT_MAP=8443:8443
+# 首次启用前先执行 ./tls/gen-cert.sh 生成证书

deploy/Dockerfile

@@ -1,23 +1,32 @@
-FROM python:3.12-slim
+FROM python:3.12-slim-bookworm
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PYTHONPATH=/app/src \
+    APP_HOST=0.0.0.0 \
+    APP_PORT=4173 \
+    APP_DB_PATH=/app/data/app.db \
+    APP_STORAGE_DIR=/app/data/files
 
 WORKDIR /app
 
-COPY requirements.txt .
-RUN pip install --no-cache-dir -r requirements.txt
+RUN groupadd --system --gid 10001 app \
+    && useradd --system --uid 10001 --gid app --home-dir /app --shell /usr/sbin/nologin app \
+    && mkdir -p /app/data \
+    && chown -R app:app /app
 
-COPY server.py ./
-COPY src ./src
-COPY web ./web
+COPY --chown=app:app requirements.txt ./
+RUN python -m pip install --no-cache-dir --requirement requirements.txt
 
-ENV PYTHONPATH=/app/src
-ENV APP_HOST=0.0.0.0
-ENV APP_PORT=4173
-ENV APP_DB_PATH=/app/data/app.db
-ENV APP_STORAGE_DIR=/app/data/files
+COPY --chown=app:app server.py ./
+COPY --chown=app:app src ./src
+COPY --chown=app:app web ./web
+
+USER app
 
 EXPOSE 4173
 
-HEALTHCHECK --interval=30s --timeout=5s --start-period=20s --retries=3 \
-  CMD python -c "import urllib.request; urllib.request.urlopen('http://127.0.0.1:4173/', timeout=4)"
+HEALTHCHECK --interval=15s --timeout=3s --start-period=10s --retries=5 \
+    CMD ["python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:4173/', timeout=2)"]
 
 CMD ["python", "server.py"]

deploy/README.md

@@ -0,0 +1,126 @@
+# 部署与运维手册（与 `release/prod` 基线一致）
+
+最后核对：2026-08-30，总工 HEL-270 上线前总验收。本手册是测试与正式环境唯一的部署/回滚/备份/巡检操作依据，取代散落在宿主机 `DEPLOYMENT.txt` 里的历史记录（旧记录仅作存档）。
+
+## 1. 物料与目录
+
+| 物料 | 说明 |
+|---|---|
+| `deploy/Dockerfile` | python:3.12-slim-bookworm，非 root（uid 10001 `app`），只读根文件系统 |
+| `deploy/compose.yaml` | cap_drop ALL、no-new-privileges、pids 128、mem 512m、cpu 1.0、healthcheck |
+| `deploy/.env.example` | 配置模板；`deploy/.env` 已 gitignore，口令永不入库 |
+| `deploy/deploy.sh` | `TAG=<tag> ./deploy.sh`，默认镜像 tag=当前提交 sha |
+| `deploy/rollback.sh` | `./rollback.sh <tag>`，只换应用不动数据 |
+| `deploy/backup.sh` | 在线热备 + 保留策略（30 天日备 + 12 个月月备） |
+| `deploy/monitor.sh` | 每 5 分钟健康 + 磁盘 ≥80% 告警，写 `ALERT.log` |
+
+数据目录：`deploy/../data/`（`app.db` + `files/`）。银行原始文件按 SHA-256 内容哈希不可变保存，任何操作不得改写。
+
+## 2. 正式配置铁律
+
+- **不得出现 `admin/admin123`**：`APP_BOOTSTRAP_ADMIN_PASSWORD` 正式环境不设值，首启由控制台读取随机生成的一次性初始口令，首登立即改密；管理员用户名避开 `admin`。
+- **登录失败限流默认开启**（5 次/10 分钟锁账号+IP）。`APP_LOGIN_RATE_LIMIT_DISABLED=1` 仅限本地调试，正式 .env 不得携带。
+- 密钥/口令/证书私钥不进仓库、不进评论、不进任务元数据。
+- **HTTPS**：正式暴露一律经反向代理（推荐同机 Caddy 容器，内网自签或内部 CA 证书，反代到容器 4173）；不占用 NAS 系统 nginx 的 80/443。公司电脑分发根证书即可。
+- **日志**：容器 stdout 走 docker json-file，限幅 `max-size=10m, max-file=3`（daemon 或 compose `logging` 配置）；审计日志在库内（append-only 触发器保护）。
+
+## 3. 部署步骤（测试与正式一致）
+
+```bash
+git fetch && git checkout release/prod && git pull
+cd deploy
+cp .env.example .env   # 首次；按上面铁律填写
+# .env 必设 APP_UID/APP_GID = data/ 目录属主的 uid/gid（宿主机 `id <用户>`），
+# 否则 SQLite WAL 写库报 readonly
+TAG=$(git rev-parse --short HEAD) ./deploy.sh
+curl -fsS http://127.0.0.1:4173/ >/dev/null && echo healthy
+```
+
+部署前手工备份（四件套 + 镜像 tag 存档）：
+
+```bash
+STAMP=$(date +%Y%m%d-%H%M%S)
+tar -C .. -czf ../backups/source-$STAMP.tar.gz --exclude='../data' --exclude='../.git' .
+cp .env ../backups/env-$STAMP
+docker exec caiwuzongzhang-app python -c "import sqlite3; s=sqlite3.connect('/app/data/app.db'); d=sqlite3.connect('/tmp/b.db'); s.backup(d); d.close(); s.close()"
+docker cp caiwuzongzhang-app:/tmp/b.db ../backups/app.db.$STAMP
+```
+
+## 3.1 正式环境部署（同宿主机隔离，2026-08-30 首次上线采用）
+
+测试与正式同宿主机并行，靠 compose 项目名/容器名/端口三隔离：
+
+| 项 | 测试环境 | 正式环境 |
+|---|---|---|
+| 目录 | `/home/leefer/caiwuzongzhang-test` | `/home/leefer/caiwuzongzhang-prod` |
+| compose 项目 | `deploy`（默认） | `COMPOSE_PROJECT_NAME=caiwuzongzhang-prod` |
+| 应用容器 | `caiwuzongzhang-app` | `caiwuzongzhang-prod-app` |
+| 端口 | `4173`（对外） | `127.0.0.1:4174`（仅本机，反代上游） |
+| HTTPS | 无 | nginx 反代 `8443`（`COMPOSE_PROFILES=tls`） |
+| 访问 | http://192.168.200.36:4173/ | https://192.168.200.36:8443/ |
+
+首次部署顺序：
+
+```bash
+git clone <repo> /home/leefer/caiwuzongzhang-prod && cd /home/leefer/caiwuzongzhang-prod
+git checkout release/prod
+mkdir -p data/files backups && chown -R 10001:10001 data   # 容器内 app uid
+cd deploy
+cp .env.example .env    # 按下述差异填写
+./tls/gen-cert.sh       # 生成内部 CA + 服务器证书（SAN 含对外 IP）
+TAG=<发布镜像tag> docker compose up -d    # 不带 --build，直接用已验收镜像
+docker logs <应用容器> 2>&1 | grep "shown once"   # 首启一次性初始口令，只出现一次
+```
+
+正式 `.env` 与测试的差异（铁律）：
+
+- `COMPOSE_PROJECT_NAME=caiwuzongzhang-prod`、`APP_CONTAINER_NAME=caiwuzongzhang-prod-app`、`APP_PORT_MAP=127.0.0.1:4174:4173`
+- `COMPOSE_PROFILES=tls`、`PROXY_CONTAINER_NAME=caiwuzongzhang-proxy`、`PROXY_PORT_MAP=8443:8443`
+- `APP_UID=10001` / `APP_GID=10001`（data/ 已 chown 10001）
+- 管理员用户名自定（如 `jinniu-admin`）；**不设** `APP_BOOTSTRAP_ADMIN_PASSWORD`（首启控制台取随机一次性口令，首登立即改密）
+- **不设** `APP_LOGIN_RATE_LIMIT_DISABLED`（限流默认开启）
+
+证书与信任分发：
+
+- `tls/certs/ca.crt` 发给各公司电脑安装到「受信任的根证书颁发机构」（安装指引见交接文档）；`ca.key`/`server.key` 永不离开宿主机。
+- 服务器证书有效期 5 年，到期前用 `./tls/gen-cert.sh` 重签并 `docker compose restart proxy`。
+
+## 4. 回滚（应用层，5 分钟内）
+
+```bash
+cd deploy && ./rollback.sh <上一个镜像tag>
+```
+
+数据层恢复见下节。两套动作互不干扰：应用回滚不改数据；数据恢复不换镜像。
+
+## 5. 备份与恢复演练
+
+- 备份：cron `15 2 * * *` 运行 `deploy/backup.sh`；月结后再手动跑一次。
+- 恢复步骤（已演练，隔离环境验证）：
+  1. `mkdir -p /tmp/restore/data && cp backups/daily/app.db.<stamp> /tmp/restore/data/app.db`
+  2. `tar -C /tmp/restore/data -xzf backups/daily/files.<stamp>.tar.gz`（解出 `files/`）
+  3. 用一份独立 `deploy-restore` 目录（`.env` 指向 `/tmp/restore`、端口错开）`docker compose up -d --build`
+  4. 核对：`PRAGMA integrity_check`、各表行数、金额合计、`source_files` 哈希清单与生产一致后才可顶替。
+- 演练频率：上线前 1 次（HEL-270 已做），之后每季度 1 次，结果记入本文件末尾。
+
+## 6. 巡检与告警
+
+cron `*/5 * * * *` 运行 `deploy/monitor.sh`；连续失败或磁盘 ≥80% 时查看 `deploy/ALERT.log` 并按需扩容/清理。容器重启策略 `unless-stopped`。
+
+宿主机 crontab（leefer，2026-08-30 起，正式环境）：
+
+```cron
+15 2 * * * cd /home/leefer/caiwuzongzhang-prod/deploy && CONTAINER=caiwuzongzhang-prod-app ./backup.sh >> backup.log 2>&1
+*/5 * * * * cd /home/leefer/caiwuzongzhang-prod/deploy && CONTAINER=caiwuzongzhang-prod-app HEALTH_URL=http://127.0.0.1:4174/ ./monitor.sh >/dev/null 2>&1
+01 3 1 * * rsync -a /home/leefer/caiwuzongzhang-prod/backups/monthly/ /vol1/caiwuzongzhang-backups/monthly/
+```
+
+（第三行为月度异盘副本：`/vol1` 是与系统盘不同的物理卷；如后续提供真正的异机目标，改为该目标。）
+
+## 7. 升级数据库
+
+应用启动时自动执行迁移（`MIGRATIONS`，当前版本 10）。迁移只前不改写历史；回退 schema 用对应 down 迁移，先备份后操作。
+
+## 8. 演练记录
+
+- 2026-08-30 HEL-270：备份恢复演练 + 版本回滚演练各 1 次，隔离环境完成，原始证据未改动。详见 HEL-270 验收评论。

deploy/backup.sh

@@ -0,0 +1,53 @@
+#!/usr/bin/env bash
+# 在线热备：不停服备份 SQLite（backup API 保证一致性）+ files/ 原始文件目录。
+# 用法：./backup.sh [备份根目录]（默认 ../backups）
+# 保留策略：日备保留 30 天，每月 1 号的首份备份额外保留 12 个月（monthly/）。
+# 建议宿主机 cron：15 2 * * * /path/to/deploy/backup.sh >> /path/to/deploy/backup.log 2>&1
+# 官方环境建议至少将 monthly/ 同步一份到异机/异盘。
+set -euo pipefail
+ROOT="$(cd "$(dirname "$0")" && pwd)"
+DEST="${1:-$ROOT/../backups}"
+CONTAINER="${CONTAINER:-caiwuzongzhang-app}"
+STAMP="$(date +%Y%m%d-%H%M%S)"
+DAY_DIR="$DEST/daily"
+MONTH_DIR="$DEST/monthly"
+mkdir -p "$DAY_DIR" "$MONTH_DIR"
+
+# 1) SQLite 在线热备（容器内 python sqlite3 backup API，主库可继续写入）
+#    备份文件写进数据卷 /app/data（宿主机 deploy/../data 可直接读取），避免
+#    read_only 容器 + tmpfs 下 docker cp 取不到文件的问题。
+docker exec -i "$CONTAINER" python - <<'PY'
+import os, sqlite3
+src = sqlite3.connect(os.environ.get("APP_DB_PATH", "/app/data/app.db"))
+dst = sqlite3.connect("/app/data/.backup-tmp.db")
+src.backup(dst)
+dst.close(); src.close()
+print("hot-backup ok")
+PY
+mv "$ROOT/../data/.backup-tmp.db" "$DAY_DIR/app.db.$STAMP"
+
+# 2) 原始文件目录打包（银行原始证据，只读复制，绝不改动）
+tar -C "$ROOT/.." -czf "$DAY_DIR/files.$STAMP.tar.gz" data/files
+
+# 3) 完整性自检：PRAGMA integrity_check + 关键计数
+docker run --rm -v "$DAY_DIR/app.db.$STAMP:/check/app.db:ro" python:3.12-slim-bookworm \
+    python -c "
+import sqlite3
+c = sqlite3.connect('/check/app.db')
+print('integrity:', c.execute('PRAGMA integrity_check').fetchone()[0])
+for t in ('source_rows', 'transfer_match_decisions', 'period_close_runs', 'audit_log'):
+    try:
+        print(t, c.execute(f'SELECT COUNT(*) FROM {t}').fetchone()[0])
+    except sqlite3.OperationalError:
+        print(t, 'n/a')
+"
+
+# 4) 保留策略：日备 >30 天删除；每月 1 号留档 monthly/
+find "$DAY_DIR" -name 'app.db.*' -mtime +30 -delete
+find "$DAY_DIR" -name 'files.*.tar.gz' -mtime +30 -delete
+if [ "$(date +%d)" = "01" ]; then
+    cp "$DAY_DIR/app.db.$STAMP" "$MONTH_DIR/app.db.$STAMP"
+    cp "$DAY_DIR/files.$STAMP.tar.gz" "$MONTH_DIR/files.$STAMP.tar.gz"
+    find "$MONTH_DIR" -mtime +365 -delete
+fi
+echo "backup complete: $DAY_DIR/app.db.$STAMP"

deploy/compose.yaml

@@ -1,36 +1,91 @@
-# 测试环境部署编排。正式口令不得写入本文件。
-# 使用：复制 .env.example 为 .env 后填写，再执行 ./deploy.sh
+# 测试/正式环境部署编排（正式口令不得写入本文件）。
+# 使用：复制 .env.example 为 .env 填写；TAG 指定镜像标签（默认取当前提交 sha）。
+# 例：TAG=v1.0.0-rc1 docker compose -f compose.yaml up -d --build
+#
+# 同宿主机多套环境（测试 + 正式）必须各自设置：
+#   COMPOSE_PROJECT_NAME / APP_CONTAINER_NAME / APP_PORT_MAP 互不相同。
+# 正式环境启用 HTTPS 反代：.env 中设 COMPOSE_PROFILES=tls（先跑 tls/gen-cert.sh）。
+
+name: ${COMPOSE_PROJECT_NAME:-deploy}
 
 services:
-  caiwuzongzhang:
+  app:
+    container_name: ${APP_CONTAINER_NAME:-caiwuzongzhang-app}
+    image: caiwuzongzhang:${TAG:-latest}
     build:
       context: ..
       dockerfile: deploy/Dockerfile
-    image: caiwuzongzhang:test
-    container_name: caiwuzongzhang-test
-    ports:
-      - "4173:4173"
     env_file:
       - .env
-    environment:
-      APP_HOST: "0.0.0.0"
-      APP_PORT: "4173"
-      APP_DB_PATH: /app/data/app.db
-      APP_STORAGE_DIR: /app/data/files
+    # 绑定挂载 data/ 时，容器运行 uid 必须与数据目录属主一致（.env 里设
+    # APP_UID/APP_GID，正式与测试环境各自填写，默认镜像内 app=10001）
+    user: "${APP_UID:-10001}:${APP_GID:-10001}"
+    ports:
+      - "${APP_PORT_MAP:-4173:4173}"
     volumes:
       - ../data:/app/data
+    init: true
+    restart: unless-stopped
     read_only: true
     tmpfs:
-      - /tmp
-    cap_drop:
-      - ALL
+      - /tmp:size=64m,mode=1777
     security_opt:
       - no-new-privileges:true
+    cap_drop:
+      - ALL
+    pids_limit: 128
     mem_limit: 512m
-    restart: unless-stopped
+    cpus: 1.0
+    logging:
+      driver: json-file
+      options:
+        max-size: "10m"
+        max-file: "3"
     healthcheck:
-      test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:4173/', timeout=4)"]
+      test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:4173/', timeout=2)"]
+      interval: 15s
+      timeout: 3s
+      retries: 5
+      start_period: 10s
+
+  # 可选 HTTPS 反向代理（正式环境用）：COMPOSE_PROFILES=tls 启用。
+  # 证书由 tls/gen-cert.sh 生成的内部 CA 签发（内网自受管信任），公司电脑
+  # 安装 tls/certs/ca.crt 一次即可无告警访问。
+  proxy:
+    container_name: ${PROXY_CONTAINER_NAME:-caiwuzongzhang-proxy}
+    image: nginx:1.27-alpine
+    profiles: ["tls"]
+    ports:
+      - "${PROXY_PORT_MAP:-8443:8443}"
+    volumes:
+      - ./tls/nginx.conf:/etc/nginx/conf.d/default.conf:ro
+      - ./tls/certs/server.crt:/etc/nginx/tls/server.crt:ro
+      - ./tls/certs/server.key:/etc/nginx/tls/server.key:ro
+    # 与证书文件属主一致的 uid/gid（默认 101=镜像 nginx 用户；宿主机生成证书时
+    # 设为属主 uid/gid，保持 server.key 0600 不放宽）
+    user: "${PROXY_UID:-101}:${PROXY_GID:-101}"
+    depends_on:
+      - app
+    restart: unless-stopped
+    read_only: true
+    tmpfs:
+      - /var/cache/nginx:size=16m,uid=${PROXY_UID:-101},gid=${PROXY_GID:-101},mode=700
+      - /var/run:size=1m,uid=${PROXY_UID:-101},gid=${PROXY_GID:-101},mode=700
+    security_opt:
+      - no-new-privileges:true
+    cap_drop:
+      - ALL
+    pids_limit: 64
+    mem_limit: 64m
+    cpus: 0.5
+    logging:
+      driver: json-file
+      options:
+        max-size: "10m"
+        max-file: "3"
+    healthcheck:
+      test: ["CMD", "wget", "-q", "--no-check-certificate", "-O", "/dev/null", "https://127.0.0.1:8443/"]
       interval: 30s
       timeout: 5s
       retries: 3
-      start_period: 20s
+      start_period: 10s

deploy/deploy.sh

@@ -1,12 +1,17 @@
 #!/usr/bin/env bash
-# 测试环境部署。施工员不执行上线；由总工在测试机运行。
+# 部署（测试或正式环境通用）。施工员不执行上线；由总工运行。
+# 用法：TAG=<镜像标签，默认当前 git 提交 sha> ./deploy.sh
 set -euo pipefail
 ROOT="$(cd "$(dirname "$0")" && pwd)"
-cd "$ROOT"
-if [[ ! -f .env ]]; then
-  echo "缺少 deploy/.env。请复制 .env.example 后填写测试口令。" >&2
-  exit 1
+cd "$ROOT/.."
+if [[ ! -f deploy/.env ]]; then
+    echo "缺少 deploy/.env。请复制 deploy/.env.example 后填写（正式口令不得入库）。" >&2
+    exit 1
 fi
-mkdir -p "$ROOT/../data/files"
-docker compose -f compose.yaml up -d --build
-echo "已启动测试环境：http://127.0.0.1:4173/"
+if [[ -z "${TAG:-}" ]]; then
+    TAG="$(git rev-parse --short HEAD 2>/dev/null || echo latest)"
+fi
+mkdir -p data/files
+cd "$ROOT"
+TAG="$TAG" docker compose -f compose.yaml up -d --build
+echo "已启动：http://127.0.0.1:${APP_PORT:-4173}/  镜像 caiwuzongzhang:$TAG"

deploy/monitor.sh

@@ -0,0 +1,32 @@
+#!/usr/bin/env bash
+# 巡检：健康端点 + 磁盘水位告警。失败/越线写 ALERT.log（供总管日检）。
+# 建议 cron：*/5 * * * * /path/to/deploy/monitor.sh >/dev/null 2>&1
+set -euo pipefail
+ROOT="$(cd "$(dirname "$0")" && pwd)"
+URL="${HEALTH_URL:-http://127.0.0.1:4173/}"
+DATA_DIR="${DATA_DIR:-$ROOT/../data}"
+ALERT="$ROOT/ALERT.log"
+DISK_THRESHOLD="${DISK_THRESHOLD:-80}"
+
+now() { date "+%Y-%m-%d %H:%M:%S"; }
+
+# 1) 健康检查：HTTP 200 才算通过
+if ! curl -fsS -m 8 -o /dev/null "$URL"; then
+    echo "[$(now)] HEALTH FAIL: $URL 无响应" >> "$ALERT"
+    exit 1
+fi
+
+# 2) 磁盘水位：数据所在分区使用率超阈值告警
+pct="$(df -P "$DATA_DIR" | awk 'NR==2 {gsub("%",""); print $5}')"
+if [ "${pct:-0}" -ge "$DISK_THRESHOLD" ]; then
+    echo "[$(now)] DISK ALERT: $DATA_DIR 使用率 ${pct}% >= ${DISK_THRESHOLD}%" >> "$ALERT"
+    exit 2
+fi
+
+# 3) 容器状态：非 running/restarting 告警
+state="$(docker inspect -f '{{.State.Status}}' "${CONTAINER:-caiwuzongzhang-app}" 2>/dev/null || echo missing)"
+if [ "$state" != "running" ]; then
+    echo "[$(now)] CONTAINER ALERT: ${CONTAINER:-caiwuzongzhang-app} state=$state" >> "$ALERT"
+    exit 3
+fi
+echo "[$(now)] ok"

deploy/rollback.sh

@@ -1,15 +1,19 @@
 #!/usr/bin/env bash
-# 测试环境回滚：停掉当前容器，按 TAG 或上一个镜像再拉起。
-# 用法：./rollback.sh [image-tag]
+# 回滚：回到指定镜像标签（或最近一次可用镜像）。
+# 用法：./rollback.sh <tag>   例：./rollback.sh f5e0915
+# 数据库与 files/ 原始证据不受影响；如需回退数据，见 deploy/README.md「数据恢复」。
 set -euo pipefail
 ROOT="$(cd "$(dirname "$0")" && pwd)"
 cd "$ROOT"
 TAG="${1:-}"
-docker compose -f compose.yaml down
-if [[ -n "$TAG" ]]; then
-  export COMPOSE_IMAGE="caiwuzongzhang:${TAG}"
-  docker compose -f compose.yaml up -d
-  echo "已回滚到镜像 caiwuzongzhang:${TAG}"
-else
-  echo "已停止测试容器。指定镜像标签可重新拉起：./rollback.sh <tag>"
+if [[ -z "$TAG" ]]; then
+    echo "用法：./rollback.sh <镜像tag>（可用 tag 见 docker images caiwuzongzhang）" >&2
+    exit 1
 fi
+if ! docker image inspect "caiwuzongzhang:$TAG" >/dev/null 2>&1; then
+    echo "镜像 caiwuzongzhang:$TAG 不存在。" >&2
+    exit 1
+fi
+TAG="$TAG" docker compose -f compose.yaml down
+TAG="$TAG" docker compose -f compose.yaml up -d
+echo "已回滚到 caiwuzongzhang:$TAG"

deploy/tls/gen-cert.sh

@@ -0,0 +1,45 @@
+#!/usr/bin/env bash
+# 生成内网专用 CA 与服务器证书（自受管内部信任，不涉及公网域名）。
+# 用法：在正式环境 deploy/ 目录执行  ./tls/gen-cert.sh
+#   可用环境变量覆盖：
+#     CERT_IP   对外访问 IP（默认 192.168.200.36）
+#     CERT_DNS  额外 DNS 名称（空格分隔，默认 caiwuzongzhang.jinniu.internal）
+# 输出：tls/certs/{ca.crt,ca.key,server.crt,server.key}（key 永不入库、不分发）
+# 分发：仅把 ca.crt 发给公司电脑安装（受信任的根证书颁发机构）。
+set -euo pipefail
+ROOT="$(cd "$(dirname "$0")" && pwd)"
+CERTS="$ROOT/certs"
+mkdir -p "$CERTS"
+cd "$CERTS"
+
+CERT_IP="${CERT_IP:-192.168.200.36}"
+CERT_DNS="${CERT_DNS:-caiwuzongzhang.jinniu.internal}"
+
+SAN="IP:${CERT_IP}"
+for d in $CERT_DNS; do SAN="$SAN,DNS:$d"; done
+
+if [[ ! -f ca.key ]]; then
+    # 内部根 CA：10 年，仅本系统使用
+    openssl req -x509 -newkey ec -pkeyopt ec_paramgen_curve:P-256 \
+        -keyout ca.key -out ca.crt -days 3650 -nodes -subj "/CN=Jinniu Ledger Internal CA" \
+        -addext "basicConstraints=critical,CA:TRUE" \
+        -addext "keyUsage=critical,keyCertSign,cRLSign"
+    chmod 600 ca.key
+    echo "ca 已生成"
+fi
+
+openssl req -newkey ec -pkeyopt ec_paramgen_curve:P-256 \
+    -keyout server.key -out server.csr -nodes \
+    -subj "/CN=${CERT_DNS%% *}" >/dev/null 2>&1
+cat > server.ext <<EXT
+subjectAltName=${SAN}
+basicConstraints=CA:FALSE
+keyUsage=digitalSignature,keyEncipherment
+extendedKeyUsage=serverAuth
+EXT
+openssl x509 -req -in server.csr -CA ca.crt -CAkey ca.key -CAcreateserial \
+    -out server.crt -days 1825 -sha256 -extfile server.ext >/dev/null 2>&1
+chmod 600 server.key
+rm -f server.csr server.ext ca.srl
+echo "server 证书已生成（SAN: ${SAN}），有效期 5 年"
+echo "分发文件：${CERTS}/ca.crt（其余文件不得离开本机）"

deploy/tls/nginx.conf

@@ -0,0 +1,31 @@
+# 正式环境 HTTPS 反代（nginx）。证书挂载自 deploy/tls/certs/（gen-cert.sh 生成）。
+# 上游为同 compose 网络内的 app:4173；对外仅暴露本代理的 8443。
+server {
+    listen 8443 ssl;
+    listen [::]:8443 ssl;
+    http2 on;
+    server_name _;
+
+    ssl_certificate     /etc/nginx/tls/server.crt;
+    ssl_certificate_key /etc/nginx/tls/server.key;
+    ssl_protocols TLSv1.2 TLSv1.3;
+    ssl_ciphers HIGH:!aNULL:!MD5;
+    ssl_prefer_server_ciphers on;
+    ssl_session_cache shared:SSL:2m;
+    ssl_session_timeout 1h;
+
+    # 银行流水 Excel 上传上限
+    client_max_body_size 25m;
+
+    proxy_http_version 1.1;
+    proxy_read_timeout 120s;
+    proxy_send_timeout 120s;
+
+    location / {
+        proxy_pass http://app:4173;
+        proxy_set_header Host $host;
+        proxy_set_header X-Real-IP $remote_addr;
+        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+        proxy_set_header X-Forwarded-Proto https;
+    }
+}

docs/任务清单.md

@@ -4,7 +4,7 @@
 
 ## 正在做
 
-- 无。HEL-269 施工交付后交总工审核测试环境部署。
+- HEL-270 上线前总验收（总工）：代码审核、测试环境部署、备份恢复与回滚演练、边界自动化证据。
 
 ## 已做完
 
@@ -20,10 +20,10 @@
 - 站内提醒与状态流转。
 - 前端移除演示流水/往来造数和 `localStorage` 业务状态。
 - 登录页视觉融合改版；月结/重开/审计按 HEL-268 视觉规范落地。
-- 测试环境部署编排收编到 `deploy/`（不在施工员职责内执行上线）。
+- 测试环境部署编排收编到 `deploy/`，含非 root 加固、备份/巡检脚本与运维手册（不在施工员职责内执行上线）。
 
 ## 还没安排
 
-- 测试环境之外的运行保障，包括 HTTPS、备份、监控和正式部署方案。覆盖正式环境必须老板明确同意。
+- 正式部署（P4）：HTTPS 反代落地、异机备份副本、正式账号交接。覆盖正式环境必须老板明确同意。
 
 每完成或新增一项任务，必须在同一次提交里把它从本清单的相应栏目移走或补上，并同步更新 `最新进度.md`。

docs/最新进度.md

@@ -14,12 +14,13 @@
 - 已实现月结：结账日到达后生成待结账任务，管理员确认后锁账并生成带 SHA-256 的月报；重开须审批，窗口到期自动恢复锁定；闭期补录进入 `period_late_arrivals`，不改已结快照；写保护拒绝锁定月的普通修改。
 - 流水列表/导出、往来查询、公司端手工记录已改为服务器真实数据；空列表显示「暂无数据」。页面不再使用 `FLOW_DEMO` / `localStorage` 业务状态。
 - 已有登录页、总账端和公司端页面。管理端「结账与期初」承接月结，「审核中心」增加重开审批，「结账与期初」之后增加「审计记录」。
-- `deploy/` 收编测试环境 Dockerfile、compose、`.env.example` 与部署/回滚脚本（4173、只读根文件系统、drop ALL、512m、healthcheck）。正式口令不进仓库。
+- `deploy/` 收编测试环境 Dockerfile、compose、`.env.example` 与部署/回滚脚本（4173、非 root、只读根文件系统、drop ALL、pids/mem/cpu 限制、healthcheck）。正式口令不进仓库。
+- 备份（在线热备 + 30 天日备/12 个月月备保留策略）、巡检（健康 + 磁盘 ≥80% 告警）脚本与《部署与运维手册》`deploy/README.md` 已入库；HTTPS 走反向代理方案已在手册写明。
 
 ## 仍未完成或不能当成已完成
 
-- 生产部署所需的 HTTPS、反向代理、备份、监控和正式运行保障尚未完成；本项目当前只允许测试环境部署。覆盖正式环境必须老板明确同意。
-- 发布基线思路是从 `f5e0915` 拉长期分支 `release/prod`，由总工审核后处理，施工员不部署。
+- 正式环境尚未部署（P4）：HTTPS 反代、异机备份副本、正式账号交接在上线时落地。覆盖正式环境必须老板明确同意。
+- 发布基线：`release/prod` 长期分支已建立，`main` 待发布时快进。
 
 ## 最近验证
 

server.py

@@ -18,7 +18,7 @@ from bank_importer import (
     manual_records, master_data, matching, multipart, period_close, personal_transit,
     positions, reminders, settings, subjects,
 )
-from bank_importer.db import connect, migrate, utc_now
+from bank_importer.db import connect, migrate, transaction, utc_now
 
 
 ROOT = Path(__file__).resolve().parent
@@ -2615,22 +2615,22 @@ class AppHandler(SimpleHTTPRequestHandler):
                 )
                 return
             try:
-                result = matching.reconcile_rows(connection, writable, actor=user)
+                with transaction(connection):
+                    result = matching.reconcile_rows(connection, writable, actor=user)
+                    ledger_events.reconcile_bank_events(connection, actor=user)
+                    result["late_arrivals"] = late
+                    auth.audit(
+                        connection, "transfer_reconcile", actor=user,
+                        target=f"rows:{len(writable)}",
+                        detail=(
+                            f"created:{result['created_events']};"
+                            f"updated:{result['updated_events']};late:{late}"
+                        ),
+                        ip=self._client_ip,
+                    )
             except Exception as exc:
                 self._send_json(500, {"status": "error", "message": f"重跑匹配失败：{exc}"})
                 return
-            try:
-                ledger_events.reconcile_bank_events(connection, actor=user)
-            except Exception as exc:
-                self._send_json(500, {"status": "error", "message": f"同步往来事件失败：{exc}"})
-                return
-            result["late_arrivals"] = late
-            auth.audit(
-                connection, "transfer_reconcile", actor=user,
-                target=f"rows:{len(writable)}",
-                detail=f"created:{result['created_events']};updated:{result['updated_events']};late:{late}",
-                ip=self._client_ip,
-            )
             self._send_json(200, {"status": "ok", "matching": result, "late_arrivals": late})
         finally:
             connection.close()
@@ -2660,19 +2660,21 @@ class AppHandler(SimpleHTTPRequestHandler):
                 return
             try:
                 period_close.assert_event_writable(connection, event_id)
-                payload = matching.apply_manual_decision(
-                    connection,
-                    event_id,
-                    action,
-                    reason=str(data.get("reason") or ""),
-                    expected_revision=expected_revision,
-                    request_key=str(data.get("request_key") or "") or None,
-                    actor=user,
-                    source_row_ids=[int(item) for item in source_row_ids]
-                    if source_row_ids
-                    else None,
-                    participant=data.get("participant"),
-                )
+                with transaction(connection):
+                    payload = matching.apply_manual_decision(
+                        connection,
+                        event_id,
+                        action,
+                        reason=str(data.get("reason") or ""),
+                        expected_revision=expected_revision,
+                        request_key=str(data.get("request_key") or "") or None,
+                        actor=user,
+                        source_row_ids=[int(item) for item in source_row_ids]
+                        if source_row_ids
+                        else None,
+                        participant=data.get("participant"),
+                    )
+                    ledger_events.reconcile_bank_events(connection, actor=user)
             except period_close.PeriodLockedError as exc:
                 self._send_json(409, {"status": "error", "message": str(exc), "year_month": exc.year_month})
                 return
@@ -2682,8 +2684,6 @@ class AppHandler(SimpleHTTPRequestHandler):
             except matching.MatchInputError as exc:
                 self._send_json(400, {"status": "error", "message": str(exc)})
                 return
-            try:
-                ledger_events.reconcile_bank_events(connection, actor=user)
             except Exception as exc:
                 self._send_json(500, {"status": "error", "message": f"同步往来事件失败：{exc}"})
                 return
@@ -2976,19 +2976,21 @@ class AppHandler(SimpleHTTPRequestHandler):
             reason = str(data.get("reason") or "").strip() or "公司端确认单边流水"
             try:
                 period_close.assert_event_writable(connection, event_id)
-                payload = matching.apply_manual_decision(
-                    connection,
-                    event_id,
-                    "assign_participant",
-                    reason=reason,
-                    expected_revision=expected_revision,
-                    request_key=request_key,
-                    actor=user,
-                    participant={
-                        "role": role,
-                        "company_id": counterparty_company_id,
-                    },
-                )
+                with transaction(connection):
+                    payload = matching.apply_manual_decision(
+                        connection,
+                        event_id,
+                        "assign_participant",
+                        reason=reason,
+                        expected_revision=expected_revision,
+                        request_key=request_key,
+                        actor=user,
+                        participant={
+                            "role": role,
+                            "company_id": counterparty_company_id,
+                        },
+                    )
+                    ledger_events.reconcile_bank_events(connection, actor=user)
             except period_close.PeriodLockedError as exc:
                 self._send_json(409, {"status": "error", "message": str(exc), "year_month": exc.year_month})
                 return
@@ -2998,8 +3000,6 @@ class AppHandler(SimpleHTTPRequestHandler):
             except matching.MatchInputError as exc:
                 self._send_json(400, {"status": "error", "message": str(exc)})
                 return
-            try:
-                ledger_events.reconcile_bank_events(connection, actor=user)
             except Exception as exc:
                 self._send_json(500, {"status": "error", "message": f"同步往来事件失败：{exc}"})
                 return
@@ -3417,51 +3417,56 @@ class AppHandler(SimpleHTTPRequestHandler):
                 period_close.assert_ledger_writable(connection, event_id)
                 if action in ("adjust", "reverse") and data.get("effective_at"):
                     period_close.assert_date_writable(connection, str(data.get("effective_at")))
-                if action == "reverse":
-                    event_id, _revision_id = ledger_events.create_reversal(
-                        connection, event_id,
-                        source_kind="adjustment",
-                        source_revision_token=None,
-                        effective_at=data.get("effective_at") or None,
-                        reason=reason, actor=user, idempotency_key=request_key,
-                    )
-                    outcome: dict[str, object] = {
-                        "action": "reverse", "ledger_event_id": event_id,
-                    }
-                elif action == "adjust":
-                    try:
-                        effective_at = str(data.get("effective_at") or "")
-                        amount = str(data.get("amount") or "")
-                        currency = str(data.get("currency") or "")
-                        payer = int(data["payer_company_id"])
-                        payee = int(data["payee_company_id"])
-                        perspective = int(data["perspective_company_id"])
-                        subject_code = str(data.get("subject_code") or "")
-                    except (KeyError, TypeError, ValueError):
-                        self._send_json(
-                            400, {"status": "error", "message": "adjust 参数不完整或无效。"}
-                        )
-                        return
-                    event_id, _revision_id = ledger_events.create_adjustment(
-                        connection, event_id,
-                        effective_at=effective_at, amount=amount, currency=currency,
-                        payer_company_id=payer, payee_company_id=payee,
-                        perspective_company_id=perspective, subject_code=subject_code,
-                        reason=reason, actor=user, idempotency_key=request_key,
-                    )
-                    outcome = {"action": "adjust", "ledger_event_id": event_id}
-                elif action == "reopen":
-                    event_id, _revision_id = ledger_events.reopen_subject(
-                        connection, event_id, reason=reason, actor=user,
-                        idempotency_key=request_key,
-                    )
-                    outcome = {"action": "reopen", "ledger_event_id": event_id}
-                else:
+                if action not in ("reverse", "adjust", "reopen"):
                     self._send_json(
                         400,
                         {"status": "error", "message": "action 必须是 reverse、adjust 或 reopen。"},
                     )
                     return
+                with transaction(connection):
+                    if action == "reverse":
+                        event_id, _revision_id = ledger_events.create_reversal(
+                            connection, event_id,
+                            source_kind="adjustment",
+                            source_revision_token=None,
+                            effective_at=data.get("effective_at") or None,
+                            reason=reason, actor=user, idempotency_key=request_key,
+                        )
+                        outcome = {
+                            "action": "reverse", "ledger_event_id": event_id,
+                        }
+                    elif action == "adjust":
+                        try:
+                            effective_at = str(data.get("effective_at") or "")
+                            amount = str(data.get("amount") or "")
+                            currency = str(data.get("currency") or "")
+                            payer = int(data["payer_company_id"])
+                            payee = int(data["payee_company_id"])
+                            perspective = int(data["perspective_company_id"])
+                            subject_code = str(data.get("subject_code") or "")
+                        except (KeyError, TypeError, ValueError):
+                            self._send_json(
+                                400, {"status": "error", "message": "adjust 参数不完整或无效。"}
+                            )
+                            return
+                        event_id, _revision_id = ledger_events.create_adjustment(
+                            connection, event_id,
+                            effective_at=effective_at, amount=amount, currency=currency,
+                            payer_company_id=payer, payee_company_id=payee,
+                            perspective_company_id=perspective, subject_code=subject_code,
+                            reason=reason, actor=user, idempotency_key=request_key,
+                        )
+                        outcome = {"action": "adjust", "ledger_event_id": event_id}
+                    else:
+                        event_id, _revision_id = ledger_events.reopen_subject(
+                            connection, event_id, reason=reason, actor=user,
+                            idempotency_key=request_key,
+                        )
+                        outcome = {"action": "reopen", "ledger_event_id": event_id}
+                    auth.audit(
+                        connection, f"ledger_{action}", actor=user,
+                        target=f"ledger_event:{event_id}", detail=reason, ip=self._client_ip,
+                    )
             except period_close.PeriodLockedError as exc:
                 self._send_json(409, {"status": "error", "message": str(exc), "year_month": exc.year_month})
                 return
@@ -3471,10 +3476,6 @@ class AppHandler(SimpleHTTPRequestHandler):
             except ledger_events.LedgerInputError as exc:
                 self._send_json(400, {"status": "error", "message": str(exc)})
                 return
-            auth.audit(
-                connection, f"ledger_{action}", actor=user,
-                target=f"ledger_event:{event_id}", detail=reason, ip=self._client_ip,
-            )
             self._send_json(200, {"status": "ok", **outcome})
         finally:
             connection.close()

src/bank_importer/auth.py

@@ -18,7 +18,7 @@ import secrets
 import sqlite3
 import string
 
-from .db import utc_now
+from .db import transaction, utc_now
 
 
 MIN_PASSWORD_LENGTH = 8
@@ -308,7 +308,7 @@ def audit(
     ip: str | None = None,
 ) -> None:
     """Append an audit log entry. Never pass passwords in ``detail``."""
-    with connection:
+    with transaction(connection):
         connection.execute(
             """
             INSERT INTO audit_log (

src/bank_importer/calculation.py

@@ -7,8 +7,8 @@ from decimal import Decimal, InvalidOperation
 import json
 import sqlite3
 
-from .db import utc_now
-from . import master_data, matching
+from .db import transaction, utc_now
+from . import auth, master_data, matching
 
 
 SETTING_START_DATE = "calculation_start_date"
@@ -81,7 +81,7 @@ def set_calculation_start_date(
         raise LockedError("已有结账月份，起算日已锁定。")
     before = get_calculation_start_date(connection)
     now = utc_now()
-    with connection:
+    with transaction(connection):
         connection.execute(
             """
             INSERT INTO system_settings (key, value, updated_at, updated_by)
@@ -185,7 +185,7 @@ def create_opening_balance(
         raise ConflictError("该对公司已有确认期初，请使用修订。")
     revision = _next_revision(connection, low_id, high_id)
     now = utc_now()
-    with connection:
+    with transaction(connection):
         cursor = connection.execute(
             """
             INSERT INTO opening_balance_revisions (
@@ -235,7 +235,7 @@ def confirm_opening_balance(
     reason = str(reason or "").strip()
     if len(reason) < 2:
         raise ValueError("确认期初必须填写原因。")
-    with connection:
+    with transaction(connection):
         connection.execute(
             """
             UPDATE opening_balance_revisions SET status = 'confirmed', reason = ?
@@ -278,7 +278,7 @@ def revise_opening_balance(
     high_id = int(row["company_id_high"])
     revision = _next_revision(connection, low_id, high_id)
     now = utc_now()
-    with connection:
+    with transaction(connection):
         connection.execute(
             "UPDATE opening_balance_revisions SET status = 'superseded' WHERE id = ?",
             (revision_id,),
@@ -333,7 +333,7 @@ def void_opening_balance(
     reason = str(reason or "").strip()
     if len(reason) < 2:
         raise ValueError("作废期初必须填写原因。")
-    with connection:
+    with transaction(connection):
         connection.execute(
             "UPDATE opening_balance_revisions SET status = 'void', reason = ? WHERE id = ?",
             (reason, revision_id),
@@ -551,7 +551,7 @@ def recalculate_coverage_gaps(connection: sqlite3.Connection) -> int:
         "SELECT * FROM bank_accounts WHERE status = 'active'"
     ).fetchall()
     rebuilt = 0
-    with connection:
+    with transaction(connection):
         for account in accounts:
             connection.execute(
                 """
@@ -668,7 +668,7 @@ def submit_no_business_attestation(
     if account["company_id"] != company_id:
         raise ValueError("只能为本公司账户提交说明。")
     now = utc_now()
-    with connection:
+    with transaction(connection):
         cursor = connection.execute(
             """
             INSERT INTO no_business_attestations (
@@ -688,6 +688,13 @@ def submit_no_business_attestation(
             ),
         )
         attestation_id = int(cursor.lastrowid)
+        auth.audit(
+            connection,
+            "attestation_submit",
+            actor=actor,
+            target=f"attestation:{attestation_id}",
+            detail=f"account:{bank_account_id};gap:{gap_start}..{gap_end}",
+        )
     return attestation_payload(connection, attestation_id)
 
 
@@ -712,7 +719,7 @@ def review_no_business_attestation(
         raise ValueError("审核必须填写理由。")
     status = "approved" if decision == "approve" else "rejected"
     now = utc_now()
-    with connection:
+    with transaction(connection):
         connection.execute(
             """
             UPDATE no_business_attestations
@@ -748,6 +755,13 @@ def review_no_business_attestation(
                 """,
                 (row["bank_account_id"], row["gap_end"], row["gap_start"]),
             )
+        auth.audit(
+            connection,
+            f"attestation_{decision}",
+            actor=actor,
+            target=f"attestation:{attestation_id}",
+            detail=review_reason,
+        )
     return attestation_payload(connection, attestation_id)
 
 

src/bank_importer/db.py

@@ -10,10 +10,12 @@ version order; each records itself in ``schema_migrations`` so re-running
 from __future__ import annotations
 
 import argparse
+from contextlib import contextmanager
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
 import sqlite3
+from typing import Iterator
 
 
 DEFAULT_DB_PATH = Path("data/app.db")
@@ -23,6 +25,31 @@ def utc_now() -> str:
     return datetime.now(timezone.utc).isoformat()
 
 
+@contextmanager
+def transaction(connection: sqlite3.Connection) -> Iterator[sqlite3.Connection]:
+    """Own a write transaction only when the caller has not already started one.
+
+    Nested helpers join the outer boundary so business rows and their audit
+    trail commit or roll back together. Standalone callers still commit before
+    return, so ``connection.close()`` cannot silently drop the work (HEL-270).
+    ``sqlite3.Connection`` as a context manager always commits on exit even
+    when it did not begin the transaction; do not use it for nestable writes.
+    """
+    began = False
+    if not connection.in_transaction:
+        connection.execute("BEGIN IMMEDIATE")
+        began = True
+    try:
+        yield connection
+    except Exception:
+        if began:
+            connection.rollback()
+        raise
+    else:
+        if began:
+            connection.commit()
+
+
 @dataclass(frozen=True)
 class Migration:
     version: int

src/bank_importer/importing.py

@@ -650,6 +650,14 @@ def review_sheets(
                 actor=actor,
             )
             ledger_events.reconcile_bank_events(connection, actor=actor)
+        if updated:
+            auth.audit(
+                connection,
+                f"sheet_{decision}",
+                actor=actor,
+                target=f"batch:{batch_id}",
+                detail=f"sheets:{','.join(updated)}" + (f";reason:{reason}" if reason else ""),
+            )
         if began:
             connection.commit()
     except Exception:
@@ -657,14 +665,6 @@ def review_sheets(
             connection.rollback()
         raise
 
-    if updated:
-        auth.audit(
-            connection,
-            f"sheet_{decision}",
-            actor=actor,
-            target=f"batch:{batch_id}",
-            detail=f"sheets:{','.join(updated)}" + (f";reason:{reason}" if reason else ""),
-        )
     payload: dict[str, object] = {"updated": updated, "already": already}
     if matching_result is not None:
         payload["matching"] = matching_result

src/bank_importer/ledger_events.py

@@ -18,7 +18,7 @@ from decimal import Decimal, InvalidOperation
 import json
 import sqlite3
 
-from .db import utc_now
+from .db import transaction, utc_now
 from .subjects import MIRROR, SUBJECTS, mirror_subject
 
 
@@ -428,26 +428,30 @@ def reopen_subject(
         raise LedgerInputError(
             "该事件没有银行来源，无法重新进入科目审核；请改用调整或冲销。"
         )
-    if not _has_reversal(connection, ledger_event_id):
-        create_reversal(
-            connection, ledger_event_id,
-            source_kind=current["source_kind"],
-            source_revision_token=current["source_revision_token"],
-            reason="科目复核：原确认事件冲销",
-            actor=actor,
-            idempotency_key=(idempotency_key + ":rev" if idempotency_key else None),
-            rule_version=current["rule_version"],
-        )
     ev = connection.execute(
         "SELECT * FROM eligible_intercompany_events WHERE event_id = ?",
         (bank_claim["bank_event_id"],),
     ).fetchone()
     if ev is None:
         raise LedgerInputError("银行事件已不再纳入往来，无法重新入账。")
-    event_id, revision_id = _create_bank_event(
-        connection, ev, actor, reason="科目复核后重新入账，待确认科目",
-        replacing_claim=bank_claim,
-    )
+    # One nestable transaction: reversal, replacement event, source re-claim
+    # and suggestions commit together. create_event used to commit on its own,
+    # leaving the bank-source UPDATE uncommitted for connection.close().
+    with transaction(connection):
+        if not _has_reversal(connection, ledger_event_id):
+            create_reversal(
+                connection, ledger_event_id,
+                source_kind=current["source_kind"],
+                source_revision_token=current["source_revision_token"],
+                reason="科目复核：原确认事件冲销",
+                actor=actor,
+                idempotency_key=(idempotency_key + ":rev" if idempotency_key else None),
+                rule_version=current["rule_version"],
+            )
+        event_id, revision_id = _create_bank_event(
+            connection, ev, actor, reason="科目复核后重新入账，待确认科目",
+            replacing_claim=bank_claim,
+        )
     return event_id, revision_id
 
 
@@ -553,51 +557,52 @@ def _create_bank_event(
     reason: str,
     replacing_claim: sqlite3.Row | None = None,
 ) -> tuple[int, int]:
-    event_id, revision_id = create_event(
-        connection,
-        state="pending_subject",
-        effective_at=event["effective_at"],
-        amount=event["amount"],
-        currency=event["currency"],
-        payer_company_id=event["payer_company_id"],
-        payee_company_id=event["payee_company_id"],
-        perspective_company_id=None,
-        subject_code=None,
-        source_kind="bank",
-        source_revision_token=event["decision_id"],
-        posting_kind="normal",
-        rule_version=SUBJECT_RULE_VERSION,
-        evidence_json=json.dumps(
-            {
-                "bank_event_id": event["event_id"],
-                "decision_id": event["decision_id"],
-                "pairing": event["pairing"],
-                "evidence_count": event["evidence_count"],
-            },
-            ensure_ascii=False,
-        ),
-        actor=actor,
-        reason=reason,
-    )
-    if replacing_claim is not None:
-        connection.execute(
-            """
-            UPDATE ledger_event_bank_sources SET ledger_event_id = ?
-            WHERE bank_event_id = ?
-            """,
-            (event_id, event["event_id"]),
+    with transaction(connection):
+        event_id, revision_id = create_event(
+            connection,
+            state="pending_subject",
+            effective_at=event["effective_at"],
+            amount=event["amount"],
+            currency=event["currency"],
+            payer_company_id=event["payer_company_id"],
+            payee_company_id=event["payee_company_id"],
+            perspective_company_id=None,
+            subject_code=None,
+            source_kind="bank",
+            source_revision_token=event["decision_id"],
+            posting_kind="normal",
+            rule_version=SUBJECT_RULE_VERSION,
+            evidence_json=json.dumps(
+                {
+                    "bank_event_id": event["event_id"],
+                    "decision_id": event["decision_id"],
+                    "pairing": event["pairing"],
+                    "evidence_count": event["evidence_count"],
+                },
+                ensure_ascii=False,
+            ),
+            actor=actor,
+            reason=reason,
         )
-    else:
-        connection.execute(
-            """
-            INSERT INTO ledger_event_bank_sources (bank_event_id, ledger_event_id)
-            VALUES (?, ?)
-            """,
-            (event["event_id"], event_id),
-        )
-    from .subjects import store_suggestions
+        if replacing_claim is not None:
+            connection.execute(
+                """
+                UPDATE ledger_event_bank_sources SET ledger_event_id = ?
+                WHERE bank_event_id = ?
+                """,
+                (event_id, event["event_id"]),
+            )
+        else:
+            connection.execute(
+                """
+                INSERT INTO ledger_event_bank_sources (bank_event_id, ledger_event_id)
+                VALUES (?, ?)
+                """,
+                (event["event_id"], event_id),
+            )
+        from .subjects import store_suggestions
 
-    store_suggestions(connection, event_id)
+        store_suggestions(connection, event_id)
     return event_id, revision_id
 
 

src/bank_importer/master_data.py

@@ -17,7 +17,7 @@ import json
 import re
 import sqlite3
 
-from .db import utc_now
+from .db import transaction, utc_now
 
 
 ACCOUNT_TYPES = ("基本户", "一般户", "专用户")
@@ -143,7 +143,7 @@ def create_company(
         raise ValueError("公司名称不能为空。")
     now = utc_now()
     try:
-        with connection:
+        with transaction(connection):
             cursor = connection.execute(
                 """
                 INSERT INTO companies (
@@ -153,16 +153,15 @@ def create_company(
                 (name, (credit_code or "").strip() or None,
                  (cashier_name or "").strip() or None, now, now),
             )
+            company_id = int(cursor.lastrowid)
+            record_change(
+                connection, "company", company_id, "create",
+                None, {"name": name, "credit_code": credit_code or None,
+                       "cashier_name": cashier_name or None, "status": "active"},
+                None, actor,
+            )
     except sqlite3.IntegrityError as exc:
         raise ConflictError("公司名称已存在。") from exc
-    company_id = int(cursor.lastrowid)
-    with connection:
-        record_change(
-            connection, "company", company_id, "create",
-            None, {"name": name, "credit_code": credit_code or None,
-                   "cashier_name": cashier_name or None, "status": "active"},
-            None, actor,
-        )
     return company_id
 
 
@@ -210,7 +209,7 @@ def submit_bank_account(
 
     if existing is None:
         try:
-            with connection:
+            with transaction(connection):
                 cursor = connection.execute(
                     """
                     INSERT INTO bank_accounts (
@@ -222,23 +221,22 @@ def submit_bank_account(
                     (company_id, number, holder, bank, kind,
                      requested_from, actor["id"] if actor else None, now, now),
                 )
+                account_id = int(cursor.lastrowid)
+                record_change(
+                    connection, "bank_account", account_id, "submit", None,
+                    {"company_id": company_id, "account_number": number,
+                     "bank_name": bank, "account_type": kind, "status": "pending",
+                     "effective_from": requested_from},
+                    None, actor,
+                )
         except sqlite3.IntegrityError as exc:
             # Lost a concurrent-insert race on the UNIQUE constraint.
             raise ConflictError("该银行账号已登记，请等待现有申请处理。") from exc
-        account_id = int(cursor.lastrowid)
-        with connection:
-            record_change(
-                connection, "bank_account", account_id, "submit", None,
-                {"company_id": company_id, "account_number": number,
-                 "bank_name": bank, "account_type": kind, "status": "pending",
-                 "effective_from": requested_from},
-                None, actor,
-            )
         return get_account(connection, account_id)
 
     if existing["status"] == "returned" and existing["company_id"] == company_id:
         before = _snapshot(existing)
-        with connection:
+        with transaction(connection):
             connection.execute(
                 """
                 UPDATE bank_accounts
@@ -287,7 +285,7 @@ def review_bank_account(
         if account["status"] != "pending":
             raise ConflictError("只有待复核的账户可以审核通过。")
         start = validate_date(effective_from, "启用日期") or account["effective_from"] or today
-        with connection:
+        with transaction(connection):
             connection.execute(
                 """
                 UPDATE bank_accounts
@@ -306,7 +304,7 @@ def review_bank_account(
             raise ConflictError("只有待复核的账户可以退回。")
         if reason is None:
             raise ValueError("退回必须填写原因。")
-        with connection:
+        with transaction(connection):
             connection.execute(
                 """
                 UPDATE bank_accounts
@@ -326,7 +324,7 @@ def review_bank_account(
         if reason is None:
             raise ValueError("停用必须填写原因。")
         end = validate_date(effective_to, "停用日期") or today
-        with connection:
+        with transaction(connection):
             connection.execute(
                 """
                 UPDATE bank_accounts
@@ -448,7 +446,7 @@ def add_alias(
     if start and end and end < start:
         raise ValueError("别名失效日期不能早于生效日期。")
     try:
-        with connection:
+        with transaction(connection):
             cursor = connection.execute(
                 """
                 INSERT INTO account_aliases (
@@ -459,17 +457,16 @@ def add_alias(
                 (account_id, alias_kind, value, rank, start, end,
                  actor["id"] if actor else None, utc_now()),
             )
+            alias_id = int(cursor.lastrowid)
+            record_change(
+                connection, "account_alias", alias_id, "create", None,
+                {"bank_account_id": account_id, "alias_kind": alias_kind,
+                 "alias_value": value, "priority": rank,
+                 "effective_from": start, "effective_to": end},
+                None, actor,
+            )
     except sqlite3.IntegrityError as exc:
         raise ConflictError("该账户下相同别名已存在。") from exc
-    alias_id = int(cursor.lastrowid)
-    with connection:
-        record_change(
-            connection, "account_alias", alias_id, "create", None,
-            {"bank_account_id": account_id, "alias_kind": alias_kind,
-             "alias_value": value, "priority": rank,
-             "effective_from": start, "effective_to": end},
-            None, actor,
-        )
     return alias_id
 
 

src/bank_importer/matching.py

@@ -33,7 +33,7 @@ import re
 import sqlite3
 
 from .auth import audit
-from .db import utc_now
+from .db import transaction, utc_now
 from .master_data import (
     is_identifiable,
     normalize_account_number,
@@ -1457,36 +1457,36 @@ def rebuild_current_projection(connection: sqlite3.Connection) -> int:
     no current pointer and no claims. Returns the number of current decisions
     rebuilt. Intended as a recovery/consistency entry point.
     """
-    connection.execute("DELETE FROM transfer_observation_claims")
-    connection.execute("DELETE FROM current_transfer_decisions")
-    events = connection.execute(
-        """
-        SELECT e.id AS event_id,
-               (SELECT d2.id FROM transfer_match_decisions d2
-                 WHERE d2.event_id = e.id
-                 ORDER BY d2.revision DESC LIMIT 1) AS latest_id
-        FROM canonical_transfer_events e
-        WHERE e.lifecycle = 'active'
-        """
-    ).fetchall()
     rebuilt = 0
-    for event in events:
-        if event["latest_id"] is None:
-            continue
-        latest = connection.execute(
-            "SELECT mode FROM transfer_match_decisions WHERE id = ?",
-            (event["latest_id"],),
-        ).fetchone()
-        if latest is None or latest["mode"] == MODE_REVERSAL:
-            continue
-        observations = connection.execute(
+    with transaction(connection):
+        connection.execute("DELETE FROM transfer_observation_claims")
+        connection.execute("DELETE FROM current_transfer_decisions")
+        events = connection.execute(
             """
-            SELECT source_row_id FROM transfer_decision_observations
-            WHERE decision_id = ? ORDER BY id
+            SELECT e.id AS event_id,
+                   (SELECT d2.id FROM transfer_match_decisions d2
+                     WHERE d2.event_id = e.id
+                     ORDER BY d2.revision DESC LIMIT 1) AS latest_id
+            FROM canonical_transfer_events e
+            WHERE e.lifecycle = 'active'
             """,
-            (event["latest_id"],),
         ).fetchall()
-        with connection:
+        for event in events:
+            if event["latest_id"] is None:
+                continue
+            latest = connection.execute(
+                "SELECT mode FROM transfer_match_decisions WHERE id = ?",
+                (event["latest_id"],),
+            ).fetchone()
+            if latest is None or latest["mode"] == MODE_REVERSAL:
+                continue
+            observations = connection.execute(
+                """
+                SELECT source_row_id FROM transfer_decision_observations
+                WHERE decision_id = ? ORDER BY id
+                """,
+                (event["latest_id"],),
+            ).fetchall()
             connection.execute(
                 """
                 INSERT OR REPLACE INTO current_transfer_decisions (event_id, decision_id)
@@ -1502,7 +1502,7 @@ def rebuild_current_projection(connection: sqlite3.Connection) -> int:
                     """,
                     (observation["source_row_id"], event["event_id"], event["latest_id"]),
                 )
-        rebuilt += 1
+            rebuilt += 1
     return rebuilt
 
 

src/bank_importer/period_close.py

@@ -16,7 +16,7 @@ import json
 import re
 import sqlite3
 
-from .db import utc_now
+from .db import transaction, utc_now
 from . import calculation, dashboard, settings as settings_mod
 
 
@@ -237,37 +237,40 @@ def record_late_arrivals(
         return 0
     now = utc_now()
     inserted = 0
-    for row_id, year_month in items:
-        existing = connection.execute(
-            "SELECT id FROM period_late_arrivals WHERE source_row_id = ?",
-            (row_id,),
-        ).fetchone()
-        if existing is not None:
-            continue
-        connection.execute(
-            """
-            INSERT INTO period_late_arrivals (
-                year_month, source_row_id, status, created_at, actor_user_id, actor_username
-            ) VALUES (?, ?, 'open', ?, ?, ?)
-            """,
-            (
+    # Nestable transaction: rows and their audit trail commit or roll back
+    # together. ``with connection:`` would commit an outer caller early.
+    with transaction(connection):
+        for row_id, year_month in items:
+            existing = connection.execute(
+                "SELECT id FROM period_late_arrivals WHERE source_row_id = ?",
+                (row_id,),
+            ).fetchone()
+            if existing is not None:
+                continue
+            connection.execute(
+                """
+                INSERT INTO period_late_arrivals (
+                    year_month, source_row_id, status, created_at, actor_user_id, actor_username
+                ) VALUES (?, ?, 'open', ?, ?, ?)
+                """,
+                (
+                    year_month,
+                    row_id,
+                    now,
+                    actor["id"] if actor is not None else None,
+                    _actor_name(actor),
+                ),
+            )
+            inserted += 1
+            _append_audit(
+                connection,
+                actor,
+                "late_arrival",
                 year_month,
-                row_id,
-                now,
-                actor["id"] if actor is not None else None,
-                _actor_name(actor),
-            ),
-        )
-        inserted += 1
-        _append_audit(
-            connection,
-            actor,
-            "late_arrival",
-            year_month,
-            object_label=f"源行 {row_id}",
-            reason="闭期后到达的流水，未改写已结账快照",
-            after={"source_row_id": row_id},
-        )
+                object_label=f"源行 {row_id}",
+                reason="闭期后到达的流水，未改写已结账快照",
+                after={"source_row_id": row_id},
+            )
     return inserted
 
 
@@ -526,7 +529,7 @@ def ensure_pending_tasks(
     first = earliest_month(connection) or last
     created: list[str] = []
     month = first
-    with connection:
+    with transaction(connection):
         while month <= last:
             existing = _current_run(connection, month)
             if existing is None:
@@ -594,7 +597,7 @@ def execute_close(
     snapshot = build_snapshot(connection, year_month)
     digest = _hash_payload(snapshot)
     now = utc_now()
-    with connection:
+    with transaction(connection):
         if current is not None and current["status"] == "pending":
             version = int(current["version"])
             report_no = _report_no(year_month, version)
@@ -661,7 +664,7 @@ def mark_close_failed(
     reason = str(reason or "").strip() or "结账失败"
     current = _current_run(connection, year_month)
     now = utc_now()
-    with connection:
+    with transaction(connection):
         if current is None:
             version = 1
             connection.execute(
@@ -723,7 +726,7 @@ def request_reopen(
     if pending is not None:
         raise PeriodConflictError("该账期已有待审批的重开申请。")
     now = utc_now()
-    with connection:
+    with transaction(connection):
         cursor = connection.execute(
             """
             INSERT INTO period_reopen_requests (
@@ -782,7 +785,7 @@ def decide_reopen(
     close_run = connection.execute(
         "SELECT * FROM period_close_runs WHERE id = ?", (row["period_close_id"],)
     ).fetchone()
-    with connection:
+    with transaction(connection):
         if approve:
             window_end = (
                 datetime.now(timezone.utc) + timedelta(days=int(row["window_days"]))
@@ -869,7 +872,7 @@ def expire_reopen_windows(
     for row in rows:
         year_month = row["year_month"]
         now = utc_now()
-        with connection:
+        with transaction(connection):
             connection.execute(
                 """
                 UPDATE period_close_runs

src/bank_importer/personal_transit.py

@@ -12,7 +12,7 @@ from __future__ import annotations
 
 import sqlite3
 
-from .db import utc_now
+from .db import transaction, utc_now
 from .master_data import (
     ConflictError,
     mask_account_number,
@@ -67,7 +67,7 @@ def submit_mapping(
     ).fetchone()
     if existing is None:
         try:
-            with connection:
+            with transaction(connection):
                 cursor = connection.execute(
                     """
                     INSERT INTO personal_transit_mappings (
@@ -81,22 +81,21 @@ def submit_mapping(
                         start, actor["id"] if actor else None, now, now,
                     ),
                 )
+                mapping_id = int(cursor.lastrowid)
+                record_change(
+                    connection, "personal_transit_mapping", mapping_id, "submit", None,
+                    {"account_number": number, "account_name": holder,
+                     "represented_company_id": represented_company_id,
+                     "allowed_direction": direction, "status": "pending",
+                     "effective_from": start},
+                    None, actor,
+                )
         except sqlite3.IntegrityError as exc:
             raise ConflictError("该个人过账账号已登记，请等待现有申请处理。") from exc
-        mapping_id = int(cursor.lastrowid)
-        with connection:
-            record_change(
-                connection, "personal_transit_mapping", mapping_id, "submit", None,
-                {"account_number": number, "account_name": holder,
-                 "represented_company_id": represented_company_id,
-                 "allowed_direction": direction, "status": "pending",
-                 "effective_from": start},
-                None, actor,
-            )
         return get_mapping(connection, mapping_id)
 
     if existing["status"] == "returned":
-        with connection:
+        with transaction(connection):
             connection.execute(
                 """
                 UPDATE personal_transit_mappings
@@ -143,7 +142,7 @@ def review_mapping(
         if mapping["status"] != "pending":
             raise ConflictError("只有待复核的映射可以审核通过。")
         start = validate_date(effective_from, "生效日期") or mapping["effective_from"] or today
-        with connection:
+        with transaction(connection):
             connection.execute(
                 """
                 UPDATE personal_transit_mappings
@@ -162,7 +161,7 @@ def review_mapping(
             raise ConflictError("只有待复核的映射可以退回。")
         if reason is None:
             raise ValueError("退回必须填写原因。")
-        with connection:
+        with transaction(connection):
             connection.execute(
                 """
                 UPDATE personal_transit_mappings
@@ -182,7 +181,7 @@ def review_mapping(
         if reason is None:
             raise ValueError("停用必须填写原因。")
         end = validate_date(effective_to, "停用日期") or today
-        with connection:
+        with transaction(connection):
             connection.execute(
                 """
                 UPDATE personal_transit_mappings

src/bank_importer/settings.py

@@ -12,7 +12,7 @@ import re
 import sqlite3
 from datetime import datetime
 
-from .db import utc_now
+from .db import utc_now, transaction
 
 
 # Defaults are applied when a key is absent; the value type is always string.
@@ -93,7 +93,7 @@ def update_settings(
     if not cleaned:
         raise ValueError("没有需要保存的设置项。")
     current = get_settings(connection)
-    with connection:
+    with transaction(connection):
         for key, new_value in cleaned.items():
             old_value = current.get(key)
             if old_value == new_value:

tests/test_calculation.py

@@ -349,6 +349,66 @@ class CoverageGapTests(CalculationBase):
         ).fetchone()
         self.assertEqual("closed_attested", closed["status"])
 
+    def test_attestation_and_audit_survive_connection_close(self) -> None:
+        """HEL-282: attestation writes used to skip the change log; review also
+        nested-committed coverage recalculation before the overlap close."""
+        from bank_importer.db import connect as db_connect
+
+        self.add_confirmed_row(
+            self.company_a,
+            account_id=self.account_a["id"],
+            own_account="6222000000000001",
+            at="2026-06-21T10:00:00",
+        )
+        calculation.recalculate_coverage_gaps(self.connection)
+        gap = self.connection.execute(
+            "SELECT * FROM coverage_gaps WHERE status = 'open'"
+        ).fetchone()
+        cashier_id = auth.create_user(
+            self.connection, "cashier-close", "CashierA123", "company", self.company_a
+        )
+        cashier = self.connection.execute(
+            "SELECT * FROM users WHERE id = ?", (cashier_id,)
+        ).fetchone()
+        att = calculation.submit_no_business_attestation(
+            self.connection,
+            company_id=self.company_a,
+            bank_account_id=self.account_a["id"],
+            gap_start=gap["gap_start"],
+            gap_end=gap["gap_end"],
+            reason="当日账户无资金往来",
+            evidence=None,
+            actor=cashier,
+        )
+        calculation.review_no_business_attestation(
+            self.connection, att["id"], "approve", "审核通过", self.admin
+        )
+        att_id = att["id"]
+        self.connection.close()
+        fresh = db_connect(self.db_path)
+        try:
+            row = fresh.execute(
+                "SELECT status FROM no_business_attestations WHERE id = ?", (att_id,)
+            ).fetchone()
+            actions = [
+                item["action"]
+                for item in fresh.execute(
+                    """
+                    SELECT action FROM audit_log
+                    WHERE action LIKE 'attestation_%'
+                    ORDER BY id
+                    """
+                ).fetchall()
+            ]
+            closed = fresh.execute(
+                "SELECT status FROM coverage_gaps WHERE id = ?", (gap["id"],)
+            ).fetchone()
+        finally:
+            fresh.close()
+        self.assertEqual("approved", row["status"])
+        self.assertEqual(["attestation_submit", "attestation_approve"], actions)
+        self.assertEqual("closed_attested", closed["status"])
+
 
 class BalanceBasisTests(CalculationBase):
     def setUp(self) -> None:

tests/test_ledger_events.py

@@ -257,6 +257,56 @@ class ProjectionTests(LedgerBase):
         self.assertEqual("confirmed", revision["state"])
         self.assertEqual("receivable", revision["subject_code"])
 
+    def test_reopen_subject_survives_connection_close(self) -> None:
+        """HEL-282: create_event used to commit the replacement event while
+        the bank-source re-claim stayed uncommitted; close() dropped the claim."""
+        from bank_importer.db import connect as db_connect
+
+        self.pair(self.company_a, self.company_b, "100.00")
+        ledger_events.reconcile_bank_events(self.connection, actor=self.admin)
+        original_id = self.ledger_events()[0]["id"]
+        subjects.confirm_subject(
+            self.connection, original_id,
+            perspective_company_id=self.company_a, subject_code="receivable",
+            reason="确认应收", expected_revision=1, request_key="k1",
+            actor=self.admin,
+        )
+        new_id, _ = ledger_events.reopen_subject(
+            self.connection, original_id,
+            reason="科目复核更正为其他应收", actor=self.admin,
+        )
+        self.connection.close()
+        fresh = db_connect(self.db_path)
+        try:
+            claim = fresh.execute(
+                "SELECT ledger_event_id FROM ledger_event_bank_sources"
+            ).fetchone()
+            new_state = fresh.execute(
+                """
+                SELECT r.state FROM current_ledger_event_revisions c
+                JOIN ledger_event_revisions r ON r.id = c.revision_id
+                WHERE c.ledger_event_id = ?
+                """,
+                (new_id,),
+            ).fetchone()
+            suggestions = fresh.execute(
+                "SELECT COUNT(*) AS n FROM ledger_subject_suggestions WHERE ledger_event_id = ?",
+                (new_id,),
+            ).fetchone()["n"]
+            reversal = fresh.execute(
+                """
+                SELECT COUNT(*) AS n FROM ledger_event_revisions
+                WHERE posting_kind = 'reversal' AND reverses_ledger_event_id = ?
+                """,
+                (original_id,),
+            ).fetchone()["n"]
+        finally:
+            fresh.close()
+        self.assertEqual(new_id, claim["ledger_event_id"])
+        self.assertEqual("pending_subject", new_state["state"])
+        self.assertGreaterEqual(suggestions, 1)
+        self.assertEqual(1, reversal)
+
 
 class SubjectSuggestionTests(LedgerBase):
     def test_mirror_mapping_is_symmetric(self) -> None:

tests/test_master_data.py

@@ -210,6 +210,40 @@ class MasterDataUnitTests(unittest.TestCase):
         )
 
 
+class MasterDataCommitTests(unittest.TestCase):
+    """File-database checks that business rows and audit share one commit."""
+
+    def setUp(self) -> None:
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp_dir.cleanup)
+        self.db_path = Path(self.temp_dir.name) / "app.db"
+        self.connection = connect(self.db_path)
+        self.addCleanup(self.connection.close)
+        migrate(self.connection)
+
+    def test_create_company_and_audit_survive_connection_close(self) -> None:
+        company_id = master_data.create_company(
+            self.connection, "丁公司", None, None, actor=None
+        )
+        self.connection.close()
+        fresh = connect(self.db_path)
+        try:
+            company = fresh.execute(
+                "SELECT name FROM companies WHERE id = ?", (company_id,)
+            ).fetchone()
+            change = fresh.execute(
+                """
+                SELECT action, entity_id FROM master_data_changes
+                WHERE entity_type = 'company'
+                """
+            ).fetchone()
+        finally:
+            fresh.close()
+        self.assertEqual("丁公司", company["name"])
+        self.assertEqual("create", change["action"])
+        self.assertEqual(company_id, change["entity_id"])
+
+
 class MasterDataApiTests(unittest.TestCase):
     """Live-server workflow tests for account registration and review."""
 

tests/test_matching.py

@@ -943,6 +943,38 @@ class ProjectionRebuildTests(MatchingBase):
         self.assertEqual(sorted(before), sorted(after))
         self.assertEqual(sorted(claims_before), sorted(claims_after))
 
+    def test_rebuild_clears_stale_projection_and_survives_close(self) -> None:
+        """HEL-282: DELETEs used to stay uncommitted when nothing was restored."""
+        from bank_importer.db import connect as db_connect
+
+        row_a = self.add_row(
+            self.company_a, own_account="6222000000000001",
+            cp_account="6222000000000002", expense="100.00",
+        )
+        row_b = self.add_row(
+            self.company_b, own_account="6222000000000002",
+            cp_account="6222000000000001", income="100.00",
+        )
+        matching.reconcile_rows(self.connection, [row_a, row_b])
+        with self.connection:
+            self.connection.execute(
+                "UPDATE canonical_transfer_events SET lifecycle = 'superseded'"
+            )
+        matching.rebuild_current_projection(self.connection)
+        self.connection.close()
+        fresh = db_connect(self.db_path)
+        try:
+            remaining = fresh.execute(
+                "SELECT COUNT(*) AS n FROM current_transfer_decisions"
+            ).fetchone()["n"]
+            claims = fresh.execute(
+                "SELECT COUNT(*) AS n FROM transfer_observation_claims"
+            ).fetchone()["n"]
+        finally:
+            fresh.close()
+        self.assertEqual(0, remaining)
+        self.assertEqual(0, claims)
+
 
 class ConcurrentReconcileTests(MatchingBase):
     def test_concurrent_reconcile_creates_one_event(self) -> None:

tests/test_period_close.py

@@ -110,6 +110,33 @@ class PeriodCloseTests(LedgerBase):
         again = period_close.close_payload(self.connection, self.MONTH)
         self.assertEqual(digest, again["snapshot_hash"])
 
+    def test_late_arrivals_survive_connection_close(self) -> None:
+        """Regression: the reconcile handler closes its connection right after
+        recording locked rows; uncommitted inserts used to vanish silently."""
+        from bank_importer.db import connect as db_connect
+
+        self._cover_month()
+        self._close()
+        late_id = self.add_row(
+            self.company_a, own_account="6222000000000001",
+            expense="12.00", at="2026-07-28T11:00:00",
+        )
+        writable, locked = period_close.split_writable_row_ids(self.connection, [late_id])
+        self.assertEqual(1, period_close.record_late_arrivals(self.connection, locked, self.admin))
+        self.connection.close()
+        fresh = db_connect(self.db_path)
+        try:
+            rows = fresh.execute(
+                "SELECT source_row_id FROM period_late_arrivals"
+            ).fetchall()
+            audits = fresh.execute(
+                "SELECT action FROM period_audit_events WHERE action = 'late_arrival'"
+            ).fetchall()
+        finally:
+            fresh.close()
+        self.assertEqual([late_id], [r["source_row_id"] for r in rows])
+        self.assertEqual(1, len(audits))
+
     def test_snapshot_row_cannot_be_updated(self) -> None:
         self._cover_month()
         self._close()
@@ -162,6 +189,42 @@ class PeriodCloseTests(LedgerBase):
         self.assertEqual(1, versions[0]["version"])
         self.assertEqual(2, versions[-1]["version"])
 
+    def test_close_and_reopen_request_survive_connection_close(self) -> None:
+        """HEL-282: monthly close / reopen request must persist with audit."""
+        from bank_importer.db import connect as db_connect
+
+        self._cover_month()
+        closed = self._close()
+        req = period_close.request_reopen(
+            self.connection, self.MONTH, self.admin,
+            reason="补录金牛煤业七月运输费并核对金额",
+        )
+        report_no = closed["report_no"]
+        request_id = req["id"]
+        self.connection.close()
+        fresh = db_connect(self.db_path)
+        try:
+            run = fresh.execute(
+                "SELECT status, report_no FROM period_close_runs WHERE year_month = ?",
+                (self.MONTH,),
+            ).fetchone()
+            reopen = fresh.execute(
+                "SELECT status FROM period_reopen_requests WHERE id = ?",
+                (request_id,),
+            ).fetchone()
+            actions = {
+                row["action"]
+                for row in fresh.execute(
+                    "SELECT action FROM period_audit_events"
+                ).fetchall()
+            }
+        finally:
+            fresh.close()
+        self.assertEqual("closed", run["status"])
+        self.assertEqual(report_no, run["report_no"])
+        self.assertEqual("pending", reopen["status"])
+        self.assertTrue({"close_execute", "reopen_request"} <= actions)
+
     def test_wal_on_file_database(self) -> None:
         mode = self.connection.execute("PRAGMA journal_mode").fetchone()[0]
         self.assertEqual("wal", str(mode).lower())

tests/test_reminders_page.py

@@ -112,12 +112,9 @@ class RemindersPageLayoutSmokeTests(unittest.TestCase):
         cls.httpd.server_close()
 
     def _open_reminders(self, page, width: int) -> None:
-        html = (WEB / "admin.html").read_text(encoding="utf-8")
+        page.route("**/app.js**", lambda route: route.abort())
         page.set_viewport_size({"width": width, "height": 900})
-        page.set_content(
-            html.replace('src="app.js?v=15"', 'src=""'),
-            base_url=self.base,
-        )
+        page.goto(f"{self.base}/admin.html")
         page.evaluate(
             """() => {
               document.querySelectorAll('.app-view').forEach((el) => {