feat(expenses): backfill historical claims into expense cases

document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/CONCEPT.md

@@ -221,6 +221,7 @@
 #### 最小事件词典
 
 - `expense_case_created`
+- `historical_claim_imported`：迁移前旧单的当前快照；只证明已纳入统一费用事件，不重建或伪造迁移前审批历史。
 - `application_generated` / `application_submitted` / `application_approved`
 - `receipt_received` / `receipt_verified` / `ocr_corrected`
 - `field_suggested` / `field_accepted` / `field_edited` / `field_rejected`
@@ -243,6 +244,9 @@
 #### 兼容策略
 
 - 迁移初期可用 shadow 事件校验现有 `ExpenseClaim` 的映射；某个状态一旦正式纳入事件模型，其业务写入与 Outbox 事件必须进入同一事务。
+- 迁移前已有 `ExpenseClaim` 通过显式维护命令写入一条 `historical_claim_imported` 快照事件：事件发生时间使用真实回填时间，原创建、发生、提交和更新时间只进入内部 payload；`history_reconstructed=false`，不得按当前状态反推并伪造历史提交、审批或付款动作。
+- 历史回填默认 dry-run，必须显式提供租户、创建时间边界和数据库目标；apply 还要求精确目标确认、迁移 head、advisory lock 和分批事务。稳定幂等键、源快照指纹、已有 Link 跳过及孤立 Event 冲突拒绝共同保证可审计重跑。
+- 历史快照事件使用 `delivery_status=suppressed`，供时间线和分析读取，但不进入实时 Outbox 投递；用户态 API 仍只返回事件白名单，不暴露回填批次、源指纹、幂等键和投递状态。
 - 旧 `ReimbursementRequest` 进入只读兼容和迁移状态，停止新增第二套业务编排。
 - 现有 `risk_flags_json` 保持读取兼容，新审批、付款、归档和关系事件写入结构化表。
 - API 新字段优先追加，不在同一阶段破坏现有前端契约。
@@ -370,6 +374,7 @@ customer_contribution_margin
 ### 后端
 
 - Expense Case 状态机、关系绑定、幂等和非法状态跃迁单元测试。
+- 历史 ExpenseClaim 回填的 dry-run 零写、租户/时间边界、快照真实性、批次回滚、冲突拒绝、重复执行和数据库目标防误连测试。
 - Business Event、AI Decision、Feedback、Outcome、Memory、Automation Policy、Savings Ledger service 单元测试。
 - 服务端会话、租户隔离、角色和动作权限的正向/越权测试。
 - 连接器事件幂等、重试、回执、失败恢复和重复支付防护测试。
@@ -395,6 +400,7 @@ customer_contribution_margin
 ### 集成
 
 - 一句话申请 → 票据归集 → 自动报销 → 预审 → 审批 → 付款事件 → 入账 → 归档端到端。
+- 持久开发库只读流式克隆 → Alembic 升级 → 历史回填 dry-run/apply/重复 apply → 服务端登录 → 旧单时间线查询 → 持久库不变验证。
 - AI 建议 → 用户修改 → 退回/通过 → 记忆候选 → 下次建议变化闭环。
 - 风险命中 → 人工确认/误报 → few-shot → 新版本回放 → Canary/回滚闭环。
 - 节省机会 → 负责人执行 → 实际结果 → 财务确认 → ROI 看板闭环。
@@ -443,6 +449,7 @@ docker exec -w /app -e SERVER_VENV_DIR=/tmp/x-financial-server-venv \
 
 - 范围过大：费用闭环、AI 学习、价值分析和商业化不能同时全量实现，需要按 P0/P1/P2 阶段交付。
 - 领域模型迁移：旧 `ReimbursementRequest`、`ExpenseClaim` 和 JSON 状态并存，必须旁路记录、小步迁移和双读校验。
+- 历史证据边界：旧单快照只表达回填时可确认的当前状态，迁移前逐节点办理过程仍以原单据和既有审计为准；后续分析不得把快照事件误当成历史审批事实。
 - 认证和租户：当前客户端身份头不适合自动化和 SaaS，多租户、记忆和高风险动作开发前必须修复。
 - 会话运维：不透明会话已经替代客户端身份头，但仍需补充定时清理、活跃会话查看/全部退出、密钥轮换策略、登录限流和企业 SSO；当前 `tenant_id` 仍是最小契约，不代表跨租户查询守卫已经完成。
 - 费用事件读取边界：用户态精简 DTO 和首批 HTTP 权限测试已完成；剩余风险是同一 URL 若已有外部客户端依赖旧内部字段会产生契约变更，且未来新增敏感 payload 字段必须继续显式进入白名单，不能恢复任意字典透传。
@@ -504,3 +511,6 @@ docker exec -w /app -e SERVER_VENV_DIR=/tmp/x-financial-server-venv \
 - 2026-07-14（迁移所有权加固）：新增统一 `schema_ownership.py`，七个运行时初始化入口只创建 legacy 表；标准启动在 Alembic upgrade 前执行只读漂移预检，revision 与 migration-owned 表集合不一致时 fail-fast，且不会自动 stamp 或修改数据库。
 - 2026-07-14（真实迁移验证）：在主应用容器连接的一次性 tmpfs PostgreSQL 17 中完成空库升级、重复升级、关键约束/索引、真实外键级联、降级到 base、legacy 哨兵保留、无版本自有表漂移拒绝和再次升级，`test_alembic_migrations.py` 4 项通过，最终 revision 为 `20260713_0002`；临时容器已自动清理，持久化开发数据库复查仍未迁移。
 - 2026-07-14（剩余边界）：当前两条 revision 只覆盖 Expense Case、Business Event 和 Auth Session，完整 legacy schema baseline 及停止其余运行时 DDL 仍未完成；本轮受影响服务回归 46 项通过，既有员工目录历史部门归一化用例仍单独失败，未混入本次迁移安全范围。
+- 2026-07-14（历史旧单接入）：新增独立 `ExpenseCaseLegacyBackfillService` 和 `backfill_legacy_expense_claim_cases.py`。命令只读取显式 `DATABASE_URL`，默认 dry-run；apply 强制目标核验、精确确认、迁移 head、advisory lock 和批次事务。每张旧单只创建一条 `historical_claim_imported` 系统快照，保留源时间和指纹、明确不重建历史，并以 `suppressed` 阻止实时投递。
+- 2026-07-14（克隆库联调）：将持久开发库以只读 `pg_dump` 流式恢复到一次性 tmpfs PostgreSQL 17，原 40 张表、4 张费用单、105 名员工、248 条预算和 62 个 Agent 资产完整保留。升级到 `20260713_0002` 后首次 dry-run 识别 4 张旧单，apply 创建 4 组 Case/Link/Event，重复 apply 创建 0 条；隔离后端完成登录、`/auth/me`、旧单时间线 200 和登出，内部回填字段未出现在 API。临时数据库和隔离进程已清理，持久库复查仍为原数据签名且没有 migration-owned 表。
+- 2026-07-14（安全复核与质量验证）：交叉审查后补齐超长租户幂等键稳定哈希、Link/Case 租户一致性、孤立 Case 冲突、URL 路由参数覆盖防护和部分批次失败进度摘要；前端将快照语义明确为“纳入时状态/节点”。容器内 65 项后端定向测试、11 项前端测试、Ruff 和 Vite 生产构建通过。全库代码体积门禁仍被本次未修改的 `RiskRuleGenerationService` 807 行存量问题阻断，未混入当前功能提交。

document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/TODO.md

@@ -77,6 +77,8 @@
   证据：`reimbursements.py`、`test_reimbursement_endpoints.py`；对抗用例修复前返回 200，修复后返回 400，且目标申请和费用事件保持不变。
 - [x] [CONCEPT: 兼容策略] 建立迁移桥接：服务启动先执行 Alembic，旧 metadata bootstrap 排除 migration-owned 表。
   证据：`server_start.sh`、`schema_ownership.py`、`migration_preflight.py`、`20260713_0001_expense_case_business_events.py`、`20260713_0002_auth_sessions.py`；容器 Shell/静态检查及一次性 PostgreSQL 完整 upgrade/downgrade/re-upgrade 通过。
+- [x] [CONCEPT: 兼容策略] 为迁移前已有 `ExpenseClaim` 提供显式、幂等且不伪造办理历史的费用事件快照回填。
+  证据：`expense_case_legacy_backfill.py`、`maintenance_database_target.py`、`backfill_legacy_expense_claim_cases.py`；默认 dry-run，apply 强制租户/截止时间、精确目标、迁移 head、advisory lock 和批次事务，事件使用真实回填时间、`history_reconstructed=false` 与 `delivery_status=suppressed`。一次性克隆库首次创建 4 组 Case/Link/Event，重复 apply 创建 0 条。
 - [ ] [CONCEPT: 兼容策略] 正式切换前以 shadow 事件校验现有 `ExpenseClaim` 映射；切换后禁止关键事件可丢弃写入。
 - [ ] [CONCEPT: 兼容策略] 制定旧 `ReimbursementRequest` 只读兼容、迁移和停止新增编排的计划。
 - [ ] [CONCEPT: 兼容策略] 把新增审批、付款、归档和关系事件移出 `risk_flags_json`，保留旧数据读取兼容。
@@ -166,9 +168,11 @@
 - [ ] [CONCEPT: 测试方案] 跑通风险反馈 → few-shot → golden case → Canary → 回滚闭环。
 - [ ] [CONCEPT: 测试方案] 跑通节省机会 → 执行 → 实现 → 财务确认 → ROI 看板闭环。
 - [x] [CONCEPT: 测试方案] 为已有 Expense Case 事件时间线补充视图模型、404 降级、详情页接入及相关响应式回归，并完成前端生产构建。
-  证据：容器内 `node --test` 定向执行 99 项通过；`npm --prefix web run build` 通过。浏览器可打开本地应用，但当前未迁移数据库无法签发有效认证凭证，真实详情页联调留待迁移后完成。
+  证据：容器内 `node --test` 定向执行 99 项通过；`npm --prefix web run build` 通过。一次性克隆迁移库上的隔离后端已完成真实登录、身份读取和旧单时间线 200 联调；持久开发库仍未迁移，因此日常本地页面仍保持兼容提示。
 - [x] [CONCEPT: 测试方案] 为 AI 申请草稿事件补充事务失败回滚、同快照幂等、同 run 多版本留痕和 Steward 重放回归。
   证据：本轮受影响后端定向回归 36 项、Expense Case 前端兼容测试 9 项和 Python `ruff --select F,I` 在容器内通过。
+- [x] [CONCEPT: 测试方案] 在现有开发数据的只读一次性克隆上验证迁移、历史回填和真实认证时间线链路。
+  证据：源库与克隆初始签名均为 40 张表、4 张费用单、105 名员工、248 条预算、62 个 Agent 资产；克隆升级后 dry-run 为 eligible=4，apply created=4，重复 apply created=0，四条事件均为 system/suppressed；登录、`/auth/me`、旧单时间线和登出均返回 200，持久库复查不变且仍无 migration-owned 表。
 - [ ] [CONCEPT: 测试方案] 补充其余前端组件、键盘操作、移动真实接口和完整浏览器关键流程验证。
 - [ ] [CONCEPT: 测试方案] 所有后端、集成和迁移测试在当前主应用容器内执行，单条命令最大超时 60s。
 - [ ] [CONCEPT: 指标与验收] 记录测试、lint、typecheck、构建、端到端和未覆盖风险证据。

server/scripts/backfill_legacy_expense_claim_cases.py

@@ -0,0 +1,532 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+import uuid
+from dataclasses import asdict
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+from sqlalchemy import create_engine, text
+from sqlalchemy.engine import Connection
+from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.orm import Session
+from sqlalchemy.pool import NullPool
+
+SERVER_DIR = Path(__file__).resolve().parents[1]
+SRC_DIR = SERVER_DIR / "src"
+if str(SRC_DIR) not in sys.path:
+    sys.path.insert(0, str(SRC_DIR))
+
+from app.db.maintenance_database_target import (  # noqa: E402
+    MaintenanceDatabaseTarget,
+    MaintenanceDatabaseTargetError,
+    validate_maintenance_database_target,
+)
+from app.db.migration_preflight import (  # noqa: E402
+    MigrationPreflightError,
+    validate_migration_state,
+)
+from app.services.expense_case_legacy_backfill import (  # noqa: E402
+    DEFAULT_BATCH_SIZE,
+    MAX_BATCH_SIZE,
+    ExpenseCaseLegacyBackfillService,
+    LegacyBackfillCursor,
+)
+
+REQUIRED_ALEMBIC_REVISION = "20260713_0002"
+EXIT_CONFIGURATION = 2
+EXIT_SAFETY = 3
+EXIT_LOCKED = 4
+EXIT_CONFLICT = 5
+EXIT_RUNTIME = 6
+
+
+class BackfillCommandError(RuntimeError):
+    def __init__(
+        self,
+        message: str,
+        *,
+        exit_code: int,
+        code: str,
+        details: dict[str, Any] | None = None,
+    ) -> None:
+        super().__init__(message)
+        self.exit_code = exit_code
+        self.code = code
+        self.details = details
+
+
+def parse_created_before(value: str) -> datetime:
+    normalized = str(value or "").strip()
+    if normalized.endswith("Z"):
+        normalized = f"{normalized[:-1]}+00:00"
+    try:
+        parsed = datetime.fromisoformat(normalized)
+    except ValueError as exc:
+        raise argparse.ArgumentTypeError(
+            "--created-before 必须是带时区的 ISO 8601 时间，例如 2026-07-14T00:00:00Z"
+        ) from exc
+    if parsed.tzinfo is None or parsed.utcoffset() is None:
+        raise argparse.ArgumentTypeError("--created-before 必须显式包含时区")
+    return parsed.astimezone(UTC)
+
+
+def positive_int(value: str) -> int:
+    try:
+        parsed = int(value)
+    except ValueError as exc:
+        raise argparse.ArgumentTypeError("必须是正整数") from exc
+    if parsed < 1:
+        raise argparse.ArgumentTypeError("必须是正整数")
+    return parsed
+
+
+def non_empty_text(value: str) -> str:
+    normalized = str(value or "").strip()
+    if not normalized:
+        raise argparse.ArgumentTypeError("不能为空")
+    return normalized
+
+
+def batch_size(value: str) -> int:
+    parsed = positive_int(value)
+    if parsed > MAX_BATCH_SIZE:
+        raise argparse.ArgumentTypeError(f"不能超过 {MAX_BATCH_SIZE}")
+    return parsed
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        description="把迁移前 ExpenseClaim 诚实地接入统一费用事件；默认只预览。",
+    )
+    mode = parser.add_mutually_exclusive_group()
+    mode.add_argument("--dry-run", action="store_true", help="只读预览（默认）。")
+    mode.add_argument("--apply", action="store_true", help="按批写入历史快照事件。")
+    parser.add_argument(
+        "--tenant-id",
+        required=True,
+        type=non_empty_text,
+        help="旧单明确归属的租户 ID。",
+    )
+    parser.add_argument(
+        "--created-before",
+        required=True,
+        type=parse_created_before,
+        help="仅处理该时刻之前创建的单据，必须包含时区。",
+    )
+    parser.add_argument("--batch-size", type=batch_size, default=DEFAULT_BATCH_SIZE)
+    parser.add_argument(
+        "--max-claims",
+        type=positive_int,
+        help="最多扫描的历史单据数，用于 canary。",
+    )
+    parser.add_argument("--sample-limit", type=positive_int, default=20)
+    parser.add_argument("--expected-host", required=True)
+    parser.add_argument("--expected-database", required=True)
+    parser.add_argument(
+        "--confirm-target",
+        help="apply 时必须精确等于解析后的 host:port/database。",
+    )
+    parser.add_argument(
+        "--allow-non-disposable-target",
+        action="store_true",
+        help="允许在非 probe 数据库 apply；仍需精确确认目标。",
+    )
+    return parser
+
+
+def _cursor_payload(cursor: LegacyBackfillCursor | None) -> dict[str, str] | None:
+    if cursor is None:
+        return None
+    return {
+        "created_at": cursor.created_at.astimezone(UTC).isoformat().replace("+00:00", "Z"),
+        "claim_id": cursor.claim_id,
+    }
+
+
+def _item_payload(item: Any) -> dict[str, str]:
+    payload = asdict(item)
+    payload["disposition"] = item.disposition.value
+    return payload
+
+
+def _base_summary(
+    *,
+    mode: str,
+    target: MaintenanceDatabaseTarget,
+    tenant_id: str,
+    created_before: datetime,
+    revision: str,
+) -> dict[str, Any]:
+    return {
+        "mode": mode,
+        "database": {
+            "target": target.exact_target,
+            "url": target.sanitized_url,
+            "disposable": target.is_disposable,
+            "revision": revision,
+        },
+        "tenant_id": tenant_id,
+        "created_before": created_before.isoformat().replace("+00:00", "Z"),
+        "inspected": 0,
+        "eligible": 0,
+        "already_linked": 0,
+        "conflicts": 0,
+        "created": 0,
+        "batches": 0,
+        "limited": False,
+        "last_cursor": None,
+        "samples": [],
+    }
+
+
+def _page_limit(*, configured: int, remaining: int | None) -> int:
+    return configured if remaining is None else min(configured, remaining)
+
+
+def _execution_progress(summary: dict[str, Any], **extra: Any) -> dict[str, Any]:
+    progress = {
+        "run_id": summary.get("run_id"),
+        "partial_commit": bool(summary.get("batches")),
+        "committed_batches": int(summary.get("batches") or 0),
+        "committed_claims": int(summary.get("inspected") or 0),
+        "created": int(summary.get("created") or 0),
+        "already_linked": int(summary.get("already_linked") or 0),
+        "last_committed_cursor": summary.get("last_cursor"),
+    }
+    progress.update(extra)
+    return progress
+
+
+def _preview(
+    session: Session,
+    *,
+    tenant_id: str,
+    created_before: datetime,
+    configured_batch_size: int,
+    max_claims: int | None,
+    sample_limit: int,
+    summary: dict[str, Any],
+) -> None:
+    service = ExpenseCaseLegacyBackfillService(
+        session,
+        tenant_id=tenant_id,
+        cutoff=created_before,
+    )
+    cursor: LegacyBackfillCursor | None = None
+    remaining = max_claims
+
+    while remaining is None or remaining > 0:
+        page = service.preview(
+            batch_size=_page_limit(configured=configured_batch_size, remaining=remaining),
+            after=cursor,
+        )
+        if not page.items:
+            break
+        summary["batches"] += 1
+        summary["inspected"] += page.inspected
+        summary["eligible"] += page.eligible
+        summary["already_linked"] += page.linked
+        summary["conflicts"] += page.conflicts
+        available_samples = max(0, sample_limit - len(summary["samples"]))
+        summary["samples"].extend(_item_payload(item) for item in page.items[:available_samples])
+        cursor = page.next_cursor
+        summary["last_cursor"] = _cursor_payload(cursor)
+        if remaining is not None:
+            remaining -= page.inspected
+        if not page.has_more:
+            break
+
+    summary["limited"] = bool(remaining == 0 and page.has_more) if "page" in locals() else False
+
+
+def _acquire_advisory_lock(connection: Connection, tenant_id: str) -> str:
+    lock_name = f"legacy-expense-case-backfill:{tenant_id}"
+    acquired = connection.scalar(
+        text("SELECT pg_try_advisory_lock(hashtextextended(:lock_name, 0))"),
+        {"lock_name": lock_name},
+    )
+    connection.commit()
+    if not acquired:
+        raise BackfillCommandError(
+            "同一租户已有历史费用事件回填任务正在运行。",
+            exit_code=EXIT_LOCKED,
+            code="advisory_lock_unavailable",
+        )
+    return lock_name
+
+
+def _release_advisory_lock(connection: Connection, lock_name: str) -> None:
+    if connection.in_transaction():
+        connection.rollback()
+    connection.execute(
+        text("SELECT pg_advisory_unlock(hashtextextended(:lock_name, 0))"),
+        {"lock_name": lock_name},
+    )
+    connection.commit()
+
+
+def _apply(
+    connection: Connection,
+    *,
+    tenant_id: str,
+    created_before: datetime,
+    configured_batch_size: int,
+    max_claims: int | None,
+    sample_limit: int,
+    summary: dict[str, Any],
+) -> None:
+    lock_name = _acquire_advisory_lock(connection, tenant_id)
+    run_id = f"historical-import-{uuid.uuid4().hex}"
+    backfilled_at = datetime.now(UTC)
+    summary["run_id"] = run_id
+    summary["backfilled_at"] = backfilled_at.isoformat().replace("+00:00", "Z")
+    cursor: LegacyBackfillCursor | None = None
+    remaining = max_claims
+
+    try:
+        with Session(bind=connection, autoflush=False, expire_on_commit=False) as session:
+            service = ExpenseCaseLegacyBackfillService(
+                session,
+                tenant_id=tenant_id,
+                cutoff=created_before,
+            )
+            while remaining is None or remaining > 0:
+                session.execute(text("SET LOCAL lock_timeout = '5s'"))
+                result = service.apply_batch(
+                    run_id=run_id,
+                    batch_size=_page_limit(
+                        configured=configured_batch_size,
+                        remaining=remaining,
+                    ),
+                    after=cursor,
+                    backfilled_at=backfilled_at,
+                )
+                if not result.items:
+                    session.rollback()
+                    break
+                if result.conflicts:
+                    session.rollback()
+                    conflict_ids = [
+                        item.claim_id
+                        for item in result.items
+                        if item.disposition.value == "conflict"
+                    ]
+                    raise BackfillCommandError(
+                        f"检测到 {result.conflicts} 个数据冲突，当前批次已回滚："
+                        f"{', '.join(conflict_ids)}",
+                        exit_code=EXIT_CONFLICT,
+                        code="legacy_claim_conflict",
+                        details=_execution_progress(
+                            summary,
+                            current_batch_conflict_ids=conflict_ids,
+                        ),
+                    )
+
+                session.commit()
+                summary["batches"] += 1
+                summary["inspected"] += result.inspected
+                summary["created"] += result.created
+                summary["already_linked"] += result.skipped_linked
+                available_samples = max(0, sample_limit - len(summary["samples"]))
+                summary["samples"].extend(
+                    _item_payload(item) for item in result.items[:available_samples]
+                )
+                cursor = result.next_cursor
+                summary["last_cursor"] = _cursor_payload(cursor)
+                if remaining is not None:
+                    remaining -= result.inspected
+                if not result.has_more:
+                    break
+
+            summary["limited"] = bool(remaining == 0 and "result" in locals() and result.has_more)
+    finally:
+        _release_advisory_lock(connection, lock_name)
+
+
+def _verify_connected_database(
+    connection: Connection,
+    *,
+    expected_database: str,
+) -> tuple[str, str]:
+    database_name, database_user = connection.execute(
+        text("SELECT current_database(), current_user")
+    ).one()
+    if str(database_name) != expected_database:
+        raise BackfillCommandError(
+            "连接后的 current_database() 与 DATABASE_URL 不一致。",
+            exit_code=EXIT_SAFETY,
+            code="connected_database_mismatch",
+        )
+    return str(database_name), str(database_user)
+
+
+def run(args: argparse.Namespace) -> dict[str, Any]:
+    database_url = os.environ.get("DATABASE_URL", "")
+    if args.apply and str(args.confirm_target or "").strip() == "":
+        raise BackfillCommandError(
+            "--apply 必须提供 --confirm-target 精确确认数据库目标。",
+            exit_code=EXIT_SAFETY,
+            code="confirm_target_required",
+        )
+    target = validate_maintenance_database_target(
+        database_url,
+        expected_host=args.expected_host,
+        expected_database=args.expected_database,
+        apply=args.apply,
+        allow_non_disposable=args.allow_non_disposable_target,
+        confirm_target=args.confirm_target,
+    )
+    engine = create_engine(
+        database_url,
+        pool_pre_ping=True,
+        poolclass=NullPool,
+    )
+    try:
+        with engine.connect() as connection:
+            connected_database, connected_user = _verify_connected_database(
+                connection,
+                expected_database=target.database,
+            )
+            state = validate_migration_state(connection)
+            if state.revision != REQUIRED_ALEMBIC_REVISION:
+                raise BackfillCommandError(
+                    "数据库必须先升级到费用事件迁移 head："
+                    f"actual={state.revision or 'unversioned/base'}; "
+                    f"required={REQUIRED_ALEMBIC_REVISION}",
+                    exit_code=EXIT_SAFETY,
+                    code="migration_revision_mismatch",
+                )
+            connection.rollback()
+
+            summary = _base_summary(
+                mode="apply" if args.apply else "dry-run",
+                target=target,
+                tenant_id=args.tenant_id,
+                created_before=args.created_before,
+                revision=state.revision,
+            )
+            summary["database"]["connected_database"] = connected_database
+            summary["database"]["connected_user"] = connected_user
+
+            with Session(bind=connection, autoflush=False) as preview_session:
+                _preview(
+                    preview_session,
+                    tenant_id=args.tenant_id,
+                    created_before=args.created_before,
+                    configured_batch_size=args.batch_size,
+                    max_claims=args.max_claims,
+                    sample_limit=args.sample_limit,
+                    summary=summary,
+                )
+                preview_session.rollback()
+
+            if not args.apply:
+                summary["would_create_cases"] = summary["eligible"]
+                summary["would_create_links"] = summary["eligible"]
+                summary["would_create_events"] = summary["eligible"]
+                return summary
+            if summary["conflicts"]:
+                raise BackfillCommandError(
+                    f"预览发现 {summary['conflicts']} 个数据冲突；未执行任何写入。",
+                    exit_code=EXIT_CONFLICT,
+                    code="legacy_claim_conflict",
+                )
+
+            preview = {
+                "inspected": summary["inspected"],
+                "eligible": summary["eligible"],
+                "already_linked": summary["already_linked"],
+                "samples": summary["samples"],
+            }
+            summary.update(
+                inspected=0,
+                eligible=preview["eligible"],
+                already_linked=0,
+                conflicts=0,
+                created=0,
+                batches=0,
+                limited=False,
+                last_cursor=None,
+                samples=[],
+                preview=preview,
+            )
+            try:
+                _apply(
+                    connection,
+                    tenant_id=args.tenant_id,
+                    created_before=args.created_before,
+                    configured_batch_size=args.batch_size,
+                    max_claims=args.max_claims,
+                    sample_limit=args.sample_limit,
+                    summary=summary,
+                )
+            except BackfillCommandError:
+                raise
+            except SQLAlchemyError as exc:
+                raise BackfillCommandError(
+                    "数据库执行失败；已提交批次不会回滚，请依据进度摘要安全重跑。",
+                    exit_code=EXIT_RUNTIME,
+                    code="database_runtime_error",
+                    details=_execution_progress(summary),
+                ) from exc
+            return summary
+    finally:
+        engine.dispose()
+
+
+def _error_payload(
+    exc: Exception,
+    *,
+    code: str,
+    details: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    payload: dict[str, Any] = {"status": "error", "code": code, "message": str(exc)}
+    if details is not None:
+        payload["details"] = details
+    return payload
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = build_parser().parse_args(argv)
+    try:
+        payload = run(args)
+    except BackfillCommandError as exc:
+        print(
+            json.dumps(
+                _error_payload(exc, code=exc.code, details=exc.details),
+                ensure_ascii=False,
+            ),
+            file=sys.stderr,
+        )
+        return exc.exit_code
+    except MaintenanceDatabaseTargetError as exc:
+        print(
+            json.dumps(_error_payload(exc, code=exc.code), ensure_ascii=False),
+            file=sys.stderr,
+        )
+        return EXIT_SAFETY
+    except MigrationPreflightError as exc:
+        print(
+            json.dumps(_error_payload(exc, code="migration_preflight_failed"), ensure_ascii=False),
+            file=sys.stderr,
+        )
+        return EXIT_SAFETY
+    except (SQLAlchemyError, OSError) as exc:
+        print(
+            json.dumps(_error_payload(exc, code="database_runtime_error"), ensure_ascii=False),
+            file=sys.stderr,
+        )
+        return EXIT_RUNTIME
+
+    print(json.dumps(payload, ensure_ascii=False, indent=2, default=str))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

server/src/app/db/maintenance_database_target.py

@@ -0,0 +1,237 @@
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from typing import Any
+
+from sqlalchemy.engine import URL, make_url
+from sqlalchemy.exc import ArgumentError
+
+DEFAULT_POSTGRESQL_PORT = 5432
+DISPOSABLE_TARGET_MARKERS = ("migration-probe", "disposable-probe")
+SENSITIVE_QUERY_KEY_PARTS = (
+    "credential",
+    "passfile",
+    "password",
+    "secret",
+    "token",
+)
+FORBIDDEN_ROUTING_QUERY_KEYS = frozenset(
+    {
+        "database",
+        "dbname",
+        "host",
+        "hostaddr",
+        "options",
+        "port",
+        "service",
+        "servicefile",
+        "user",
+        "username",
+    }
+)
+
+
+class MaintenanceDatabaseTargetError(ValueError):
+    """维护命令数据库目标不满足安全约束。"""
+
+    def __init__(self, code: str, message: str) -> None:
+        super().__init__(message)
+        self.code = code
+
+
+@dataclass(frozen=True, slots=True)
+class MaintenanceDatabaseTarget:
+    host: str
+    port: int
+    database: str
+    username: str | None
+    sanitized_url: str
+    exact_target: str
+    is_disposable: bool
+
+
+def _raise_target_error(code: str, message: str) -> None:
+    raise MaintenanceDatabaseTargetError(code, message)
+
+
+def _normalize_host(value: str) -> str:
+    normalized = str(value or "").strip().lower()
+    if normalized.startswith("[") and normalized.endswith("]"):
+        normalized = normalized[1:-1]
+    return normalized.rstrip(".")
+
+
+def _normalize_probe_component(value: str) -> str:
+    return re.sub(r"[^a-z0-9]+", "-", str(value or "").lower()).strip("-")
+
+
+def _matches_disposable_marker(value: str) -> bool:
+    normalized = _normalize_probe_component(value)
+    return any(
+        normalized == marker
+        or normalized.startswith(f"{marker}-")
+        or normalized.startswith(f"x-financial-{marker}-")
+        for marker in DISPOSABLE_TARGET_MARKERS
+    )
+
+
+def is_disposable_maintenance_target(*, host: str, database: str) -> bool:
+    """只有主机名和数据库名都带显式 probe 标记时才视为一次性目标。"""
+
+    return _matches_disposable_marker(host) and _matches_disposable_marker(database)
+
+
+def _sanitize_query(url: URL) -> URL:
+    sanitized_query: dict[str, Any] = {}
+    for key, value in url.query.items():
+        normalized_key = str(key).lower()
+        if any(part in normalized_key for part in SENSITIVE_QUERY_KEY_PARTS):
+            if isinstance(value, tuple):
+                sanitized_query[key] = tuple("***" for _ in value)
+            else:
+                sanitized_query[key] = "***"
+            continue
+        sanitized_query[key] = value
+    return url.set(query=sanitized_query)
+
+
+def _format_target_host(host: str) -> str:
+    return f"[{host}]" if ":" in host else host
+
+
+def parse_maintenance_database_target(
+    database_url: str,
+    *,
+    expected_host: str,
+    expected_database: str,
+) -> MaintenanceDatabaseTarget:
+    """解析显式数据库 URL，并核对操作人声明的目标主机和数据库。"""
+
+    raw_url = str(database_url or "").strip()
+    if not raw_url:
+        _raise_target_error(
+            "database_url_required",
+            "维护命令必须显式提供 DATABASE_URL，禁止回退到环境文件或默认配置。",
+        )
+
+    try:
+        parsed_url = make_url(raw_url)
+    except (ArgumentError, TypeError, ValueError) as exc:
+        raise MaintenanceDatabaseTargetError(
+            "invalid_database_url",
+            "DATABASE_URL 不是有效的 SQLAlchemy 数据库 URL。",
+        ) from exc
+
+    if parsed_url.get_backend_name() != "postgresql":
+        _raise_target_error(
+            "postgresql_required",
+            "维护命令只允许连接 PostgreSQL 数据库。",
+        )
+
+    routing_query_keys = sorted(
+        str(key).lower()
+        for key in parsed_url.query
+        if str(key).lower() in FORBIDDEN_ROUTING_QUERY_KEYS
+    )
+    if routing_query_keys:
+        _raise_target_error(
+            "database_routing_query_forbidden",
+            f"DATABASE_URL 查询参数不得覆盖连接目标或 schema：{', '.join(routing_query_keys)}",
+        )
+
+    host = _normalize_host(parsed_url.host or "")
+    if not host:
+        _raise_target_error(
+            "database_host_required",
+            "DATABASE_URL 必须包含显式 PostgreSQL 主机名。",
+        )
+
+    database = str(parsed_url.database or "").strip()
+    if not database:
+        _raise_target_error(
+            "database_name_required",
+            "DATABASE_URL 必须包含显式数据库名。",
+        )
+
+    normalized_expected_host = _normalize_host(expected_host)
+    if not normalized_expected_host:
+        _raise_target_error(
+            "expected_host_required",
+            "必须通过 expected_host 声明预期数据库主机。",
+        )
+    if host != normalized_expected_host:
+        _raise_target_error(
+            "expected_host_mismatch",
+            f"DATABASE_URL 主机与预期不一致：actual={host}; expected={normalized_expected_host}",
+        )
+
+    normalized_expected_database = str(expected_database or "").strip()
+    if not normalized_expected_database:
+        _raise_target_error(
+            "expected_database_required",
+            "必须通过 expected_database 声明预期数据库名。",
+        )
+    if database != normalized_expected_database:
+        _raise_target_error(
+            "expected_database_mismatch",
+            "DATABASE_URL 数据库名与预期不一致："
+            f"actual={database}; expected={normalized_expected_database}",
+        )
+
+    port = int(parsed_url.port or DEFAULT_POSTGRESQL_PORT)
+    exact_target = f"{_format_target_host(host)}:{port}/{database}"
+    sanitized_url = _sanitize_query(parsed_url).render_as_string(hide_password=True)
+
+    return MaintenanceDatabaseTarget(
+        host=host,
+        port=port,
+        database=database,
+        username=parsed_url.username,
+        sanitized_url=sanitized_url,
+        exact_target=exact_target,
+        is_disposable=is_disposable_maintenance_target(host=host, database=database),
+    )
+
+
+def validate_maintenance_database_target(
+    database_url: str,
+    *,
+    expected_host: str,
+    expected_database: str,
+    apply: bool = False,
+    allow_non_disposable: bool = False,
+    confirm_target: str | None = None,
+) -> MaintenanceDatabaseTarget:
+    """校验维护目标；非一次性数据库 apply 必须显式放行并精确确认目标。"""
+
+    target = parse_maintenance_database_target(
+        database_url,
+        expected_host=expected_host,
+        expected_database=expected_database,
+    )
+    if not apply:
+        return target
+
+    normalized_confirmation = str(confirm_target or "").strip()
+    if target.is_disposable:
+        if normalized_confirmation and normalized_confirmation != target.exact_target:
+            _raise_target_error(
+                "confirm_target_mismatch",
+                "confirm_target 与解析后的数据库目标不一致："
+                f"actual={normalized_confirmation}; expected={target.exact_target}",
+            )
+        return target
+
+    if not allow_non_disposable:
+        _raise_target_error(
+            "non_disposable_apply_forbidden",
+            "非一次性数据库 apply 必须显式启用 allow_non_disposable。",
+        )
+    if normalized_confirmation != target.exact_target:
+        _raise_target_error(
+            "confirm_target_mismatch",
+            "非一次性数据库 apply 必须通过 confirm_target 精确确认："
+            f"expected={target.exact_target}",
+        )
+    return target

server/src/app/services/expense_case_legacy_backfill.py

@@ -0,0 +1,391 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from decimal import Decimal
+from enum import StrEnum
+
+from sqlalchemy import and_, or_, select
+from sqlalchemy.orm import Session
+
+from app.models.expense_case import BusinessEvent, ExpenseCase, ExpenseCaseLink
+from app.models.financial_record import ExpenseClaim
+from app.services.expense_cases import ExpenseCaseService
+
+HISTORICAL_CLAIM_IMPORTED_EVENT = "historical_claim_imported"
+HISTORICAL_IMPORT_VERSION = 1
+HISTORICAL_IMPORT_DELIVERY_STATUS = "suppressed"
+DEFAULT_BATCH_SIZE = 100
+MAX_BATCH_SIZE = 1000
+
+
+class LegacyBackfillDisposition(StrEnum):
+    ELIGIBLE = "eligible"
+    LINKED = "linked"
+    CONFLICT = "conflict"
+
+
+@dataclass(frozen=True, slots=True)
+class LegacyBackfillCursor:
+    created_at: datetime
+    claim_id: str
+
+
+@dataclass(frozen=True, slots=True)
+class LegacyBackfillItem:
+    claim_id: str
+    claim_no: str
+    disposition: LegacyBackfillDisposition
+    source_fingerprint: str
+    reason: str = ""
+
+
+@dataclass(frozen=True, slots=True)
+class LegacyBackfillPreview:
+    tenant_id: str
+    cutoff: datetime
+    inspected: int
+    eligible: int
+    linked: int
+    conflicts: int
+    has_more: bool
+    next_cursor: LegacyBackfillCursor | None
+    items: tuple[LegacyBackfillItem, ...]
+
+
+@dataclass(frozen=True, slots=True)
+class LegacyBackfillBatchResult:
+    tenant_id: str
+    cutoff: datetime
+    run_id: str
+    inspected: int
+    created: int
+    skipped_linked: int
+    conflicts: int
+    has_more: bool
+    next_cursor: LegacyBackfillCursor | None
+    items: tuple[LegacyBackfillItem, ...]
+
+
+class ExpenseCaseLegacyBackfillService:
+    """把旧费用单诚实地接入费用事件，不虚构迁移前的逐节点历史。"""
+
+    def __init__(self, db: Session, *, tenant_id: str, cutoff: datetime) -> None:
+        self.db = db
+        self.tenant_id = self._require_text(tenant_id, field_name="tenant_id", max_length=64)
+        self.cutoff = self._require_aware_datetime(cutoff, field_name="cutoff")
+        self.expense_cases = ExpenseCaseService(db)
+
+    def preview(
+        self,
+        *,
+        batch_size: int = DEFAULT_BATCH_SIZE,
+        after: LegacyBackfillCursor | None = None,
+    ) -> LegacyBackfillPreview:
+        """预览一批旧单；该方法只查询，不 flush、不 commit。"""
+
+        with self.db.no_autoflush:
+            claims, has_more = self._load_claims(
+                batch_size=self._normalize_batch_size(batch_size),
+                after=after,
+                lock_rows=False,
+            )
+            items = self._classify(claims)
+        return LegacyBackfillPreview(
+            tenant_id=self.tenant_id,
+            cutoff=self.cutoff,
+            inspected=len(items),
+            eligible=self._count(items, LegacyBackfillDisposition.ELIGIBLE),
+            linked=self._count(items, LegacyBackfillDisposition.LINKED),
+            conflicts=self._count(items, LegacyBackfillDisposition.CONFLICT),
+            has_more=has_more,
+            next_cursor=self._next_cursor(claims),
+            items=items,
+        )
+
+    def apply_batch(
+        self,
+        *,
+        run_id: str,
+        batch_size: int = DEFAULT_BATCH_SIZE,
+        after: LegacyBackfillCursor | None = None,
+        backfilled_at: datetime | None = None,
+    ) -> LegacyBackfillBatchResult:
+        """应用一批回填但不提交；调用方拥有完整的批次事务边界。"""
+
+        normalized_run_id = self._require_text(run_id, field_name="run_id", max_length=64)
+        normalized_backfilled_at = self._require_aware_datetime(
+            backfilled_at or datetime.now(UTC),
+            field_name="backfilled_at",
+        )
+        claims, has_more = self._load_claims(
+            batch_size=self._normalize_batch_size(batch_size),
+            after=after,
+            lock_rows=True,
+        )
+        classified = self._classify(claims)
+        claims_by_id = {claim.id: claim for claim in claims}
+        result_items: list[LegacyBackfillItem] = []
+        created = 0
+
+        for item in classified:
+            if item.disposition is not LegacyBackfillDisposition.ELIGIBLE:
+                result_items.append(item)
+                continue
+
+            claim = claims_by_id[item.claim_id]
+            _expense_case, event = self.expense_cases.record_claim_event(
+                claim,
+                event_type=HISTORICAL_CLAIM_IMPORTED_EVENT,
+                actor_id="system",
+                tenant_id=self.tenant_id,
+                correlation_id=normalized_run_id,
+                idempotency_key=self.idempotency_key(claim.id),
+                previous_status="",
+                previous_approval_stage="",
+                extra_payload=self._event_payload(
+                    claim,
+                    run_id=normalized_run_id,
+                    backfilled_at=normalized_backfilled_at,
+                    source_fingerprint=item.source_fingerprint,
+                ),
+                delivery_status=HISTORICAL_IMPORT_DELIVERY_STATUS,
+            )
+            # 事件发生时间表达真实回填动作；旧单业务时间只保留在 payload 中。
+            event.occurred_at = normalized_backfilled_at
+            created += 1
+            result_items.append(item)
+
+        return LegacyBackfillBatchResult(
+            tenant_id=self.tenant_id,
+            cutoff=self.cutoff,
+            run_id=normalized_run_id,
+            inspected=len(result_items),
+            created=created,
+            skipped_linked=self._count(result_items, LegacyBackfillDisposition.LINKED),
+            conflicts=self._count(result_items, LegacyBackfillDisposition.CONFLICT),
+            has_more=has_more,
+            next_cursor=self._next_cursor(claims),
+            items=tuple(result_items),
+        )
+
+    def idempotency_key(self, claim_id: str) -> str:
+        normalized_claim_id = str(claim_id or "").strip()
+        if not normalized_claim_id:
+            raise ValueError("claim_id must not be empty")
+        key = (
+            f"historical-import:v{HISTORICAL_IMPORT_VERSION}:{self.tenant_id}:{normalized_claim_id}"
+        )
+        if len(key) <= 120:
+            return key
+        digest = hashlib.sha256(key.encode("utf-8")).hexdigest()
+        return f"historical-import:v{HISTORICAL_IMPORT_VERSION}:sha256:{digest}"
+
+    @classmethod
+    def source_fingerprint(cls, claim: ExpenseClaim) -> str:
+        encoded = json.dumps(
+            cls._source_snapshot(claim),
+            ensure_ascii=False,
+            sort_keys=True,
+            separators=(",", ":"),
+        ).encode("utf-8")
+        return f"sha256:{hashlib.sha256(encoded).hexdigest()}"
+
+    def _load_claims(
+        self,
+        *,
+        batch_size: int,
+        after: LegacyBackfillCursor | None,
+        lock_rows: bool,
+    ) -> tuple[list[ExpenseClaim], bool]:
+        stmt = select(ExpenseClaim).where(ExpenseClaim.created_at < self.cutoff)
+        if after is not None:
+            normalized_claim_id = self._require_text(
+                after.claim_id,
+                field_name="after.claim_id",
+                max_length=36,
+            )
+            stmt = stmt.where(
+                or_(
+                    ExpenseClaim.created_at > after.created_at,
+                    and_(
+                        ExpenseClaim.created_at == after.created_at,
+                        ExpenseClaim.id > normalized_claim_id,
+                    ),
+                )
+            )
+        stmt = stmt.order_by(ExpenseClaim.created_at.asc(), ExpenseClaim.id.asc()).limit(
+            batch_size + 1
+        )
+        if lock_rows:
+            stmt = stmt.with_for_update()
+
+        claims = list(self.db.scalars(stmt).all())
+        return claims[:batch_size], len(claims) > batch_size
+
+    def _classify(self, claims: list[ExpenseClaim]) -> tuple[LegacyBackfillItem, ...]:
+        if not claims:
+            return ()
+
+        claim_ids = [claim.id for claim in claims]
+        links_by_claim_id = {
+            link.resource_id: (link, expense_case)
+            for link, expense_case in self.db.execute(
+                select(ExpenseCaseLink, ExpenseCase)
+                .outerjoin(ExpenseCase, ExpenseCase.id == ExpenseCaseLink.expense_case_id)
+                .where(
+                    ExpenseCaseLink.resource_type == "expense_claim",
+                    ExpenseCaseLink.resource_id.in_(claim_ids),
+                )
+            ).all()
+        }
+        event_claim_ids = set(
+            self.db.scalars(
+                select(BusinessEvent.aggregate_id).where(
+                    BusinessEvent.aggregate_type == "expense_claim",
+                    BusinessEvent.aggregate_id.in_(claim_ids),
+                )
+            ).all()
+        )
+        expected_case_nos = {f"CASE-{str(claim.claim_no or claim.id).strip()}" for claim in claims}
+        existing_case_nos = set(
+            self.db.scalars(
+                select(ExpenseCase.case_no).where(ExpenseCase.case_no.in_(expected_case_nos))
+            ).all()
+        )
+
+        items: list[LegacyBackfillItem] = []
+        for claim in claims:
+            fingerprint = self.source_fingerprint(claim)
+            linked_record = links_by_claim_id.get(claim.id)
+            expected_case_no = f"CASE-{str(claim.claim_no or claim.id).strip()}"
+            if linked_record is not None:
+                link, expense_case = linked_record
+                if link.tenant_id == self.tenant_id:
+                    if expense_case is None or expense_case.tenant_id != self.tenant_id:
+                        disposition = LegacyBackfillDisposition.CONFLICT
+                        reason = "expense claim link points to an invalid tenant expense case"
+                    else:
+                        disposition = LegacyBackfillDisposition.LINKED
+                        reason = "expense claim already belongs to an expense case"
+                else:
+                    disposition = LegacyBackfillDisposition.CONFLICT
+                    reason = "expense claim is linked under another tenant"
+            elif claim.id in event_claim_ids:
+                disposition = LegacyBackfillDisposition.CONFLICT
+                reason = "business event exists without an expense case link"
+            elif expected_case_no in existing_case_nos:
+                disposition = LegacyBackfillDisposition.CONFLICT
+                reason = "expense case exists without an expense claim link"
+            else:
+                disposition = LegacyBackfillDisposition.ELIGIBLE
+                reason = ""
+            items.append(
+                LegacyBackfillItem(
+                    claim_id=claim.id,
+                    claim_no=str(claim.claim_no or ""),
+                    disposition=disposition,
+                    source_fingerprint=fingerprint,
+                    reason=reason,
+                )
+            )
+        return tuple(items)
+
+    @classmethod
+    def _source_snapshot(cls, claim: ExpenseClaim) -> dict[str, object]:
+        return {
+            "id": str(claim.id or ""),
+            "claim_no": str(claim.claim_no or ""),
+            "employee_id": str(claim.employee_id or ""),
+            "expense_type": str(claim.expense_type or ""),
+            "amount": cls._money_text(claim.amount),
+            "currency": str(claim.currency or "CNY"),
+            "status": str(claim.status or ""),
+            "approval_stage": str(claim.approval_stage or ""),
+            "source_times": cls._source_times(claim),
+        }
+
+    @classmethod
+    def _event_payload(
+        cls,
+        claim: ExpenseClaim,
+        *,
+        run_id: str,
+        backfilled_at: datetime,
+        source_fingerprint: str,
+    ) -> dict[str, object]:
+        return {
+            "schema_version": HISTORICAL_IMPORT_VERSION,
+            "backfill_version": HISTORICAL_IMPORT_VERSION,
+            "source": "legacy_expense_claim",
+            "history_reconstructed": False,
+            "backfill_run_id": run_id,
+            "backfilled_at": cls._isoformat(backfilled_at),
+            "performed_by": "expense_case_legacy_backfill_cli",
+            "source_times": cls._source_times(claim),
+            "source_fingerprint": source_fingerprint,
+            "reason": "该单据已纳入统一费用事件；迁移前的逐节点办理明细未重建。",
+        }
+
+    @classmethod
+    def _source_times(cls, claim: ExpenseClaim) -> dict[str, str | None]:
+        return {
+            "occurred_at": cls._optional_isoformat(claim.occurred_at),
+            "submitted_at": cls._optional_isoformat(claim.submitted_at),
+            "created_at": cls._optional_isoformat(claim.created_at),
+            "updated_at": cls._optional_isoformat(claim.updated_at),
+        }
+
+    @staticmethod
+    def _money_text(value: Decimal | None) -> str:
+        return f"{Decimal(value or Decimal('0.00')).quantize(Decimal('0.01')):.2f}"
+
+    @classmethod
+    def _optional_isoformat(cls, value: datetime | None) -> str | None:
+        return cls._isoformat(value) if value is not None else None
+
+    @staticmethod
+    def _isoformat(value: datetime) -> str:
+        normalized = value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)
+        return normalized.isoformat().replace("+00:00", "Z")
+
+    @staticmethod
+    def _require_aware_datetime(value: datetime, *, field_name: str) -> datetime:
+        if value.tzinfo is None or value.utcoffset() is None:
+            raise ValueError(f"{field_name} must include a timezone")
+        return value.astimezone(UTC)
+
+    @staticmethod
+    def _require_text(value: str, *, field_name: str, max_length: int) -> str:
+        normalized = str(value or "").strip()
+        if not normalized:
+            raise ValueError(f"{field_name} must not be empty")
+        if len(normalized) > max_length:
+            raise ValueError(f"{field_name} must be at most {max_length} characters")
+        return normalized
+
+    @staticmethod
+    def _normalize_batch_size(value: int) -> int:
+        normalized = int(value)
+        if normalized < 1 or normalized > MAX_BATCH_SIZE:
+            raise ValueError(f"batch_size must be between 1 and {MAX_BATCH_SIZE}")
+        return normalized
+
+    @staticmethod
+    def _count(
+        items: tuple[LegacyBackfillItem, ...] | list[LegacyBackfillItem],
+        disposition: LegacyBackfillDisposition,
+    ) -> int:
+        return sum(1 for item in items if item.disposition is disposition)
+
+    @staticmethod
+    def _next_cursor(claims: list[ExpenseClaim]) -> LegacyBackfillCursor | None:
+        if not claims:
+            return None
+        last_claim = claims[-1]
+        return LegacyBackfillCursor(
+            created_at=last_claim.created_at,
+            claim_id=last_claim.id,
+        )

server/src/app/services/expense_cases.py

@@ -151,6 +151,7 @@ class ExpenseCaseService:
         expense_case: ExpenseCase | None = None,
         relation_type: str | None = None,
         update_case_state: bool = True,
+        delivery_status: str = "pending",
     ) -> tuple[ExpenseCase, BusinessEvent]:
         normalized_tenant = self.normalize_tenant_id(tenant_id)
         if expense_case is None:
@@ -215,7 +216,7 @@ class ExpenseCaseService:
             actor_id=str(actor_id or "system").strip() or "system",
             actor_type="system" if str(actor_id or "").strip() == "system" else "user",
             payload_json=payload,
-            delivery_status="pending",
+            delivery_status=str(delivery_status or "pending").strip() or "pending",
             occurred_at=datetime.now(UTC),
         )
         self.db.add(event)

server/tests/test_backfill_legacy_expense_claim_cases_cli.py

@@ -0,0 +1,183 @@
+from __future__ import annotations
+
+import argparse
+import importlib.util
+from datetime import UTC, datetime
+from pathlib import Path
+
+import pytest
+
+
+def load_script_module():
+    script_path = (
+        Path(__file__).resolve().parents[1] / "scripts" / "backfill_legacy_expense_claim_cases.py"
+    )
+    spec = importlib.util.spec_from_file_location(
+        "backfill_legacy_expense_claim_cases_cli",
+        script_path,
+    )
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+MODULE = load_script_module()
+
+
+def test_created_before_requires_timezone_and_normalizes_to_utc() -> None:
+    assert MODULE.parse_created_before("2026-07-14T08:00:00+08:00") == datetime(
+        2026,
+        7,
+        14,
+        tzinfo=UTC,
+    )
+    with pytest.raises(argparse.ArgumentTypeError, match="时区"):
+        MODULE.parse_created_before("2026-07-14T00:00:00")
+
+
+@pytest.mark.parametrize("value", ["0", "-1", "not-a-number"])
+def test_positive_int_rejects_invalid_values(value: str) -> None:
+    with pytest.raises(argparse.ArgumentTypeError):
+        MODULE.positive_int(value)
+
+
+def test_parser_defaults_to_dry_run_and_requires_explicit_scope() -> None:
+    args = MODULE.build_parser().parse_args(
+        [
+            "--tenant-id",
+            "default",
+            "--created-before",
+            "2026-07-14T00:00:00Z",
+            "--expected-host",
+            "migration-probe",
+            "--expected-database",
+            "migration_probe",
+        ]
+    )
+
+    assert args.apply is False
+    assert args.dry_run is False
+    assert args.batch_size == 100
+    assert args.tenant_id == "default"
+
+
+def test_parser_rejects_apply_and_dry_run_together() -> None:
+    with pytest.raises(SystemExit):
+        MODULE.build_parser().parse_args(
+            [
+                "--apply",
+                "--dry-run",
+                "--tenant-id",
+                "default",
+                "--created-before",
+                "2026-07-14T00:00:00Z",
+                "--expected-host",
+                "migration-probe",
+                "--expected-database",
+                "migration_probe",
+            ]
+        )
+
+
+def test_parser_rejects_blank_tenant() -> None:
+    with pytest.raises(SystemExit):
+        MODULE.build_parser().parse_args(
+            [
+                "--tenant-id",
+                " ",
+                "--created-before",
+                "2026-07-14T00:00:00Z",
+                "--expected-host",
+                "migration-probe",
+                "--expected-database",
+                "migration_probe",
+            ]
+        )
+
+
+def test_apply_requires_exact_target_confirmation_before_connecting(monkeypatch) -> None:
+    monkeypatch.setenv(
+        "DATABASE_URL",
+        "postgresql://finance:secret@migration-probe/migration_probe",
+    )
+    args = MODULE.build_parser().parse_args(
+        [
+            "--apply",
+            "--tenant-id",
+            "default",
+            "--created-before",
+            "2026-07-14T00:00:00Z",
+            "--expected-host",
+            "migration-probe",
+            "--expected-database",
+            "migration_probe",
+        ]
+    )
+
+    with pytest.raises(MODULE.BackfillCommandError) as exc_info:
+        MODULE.run(args)
+
+    assert exc_info.value.code == "confirm_target_required"
+    assert exc_info.value.exit_code == MODULE.EXIT_SAFETY
+
+
+def test_missing_database_url_is_rejected_without_configuration_fallback(monkeypatch) -> None:
+    monkeypatch.delenv("DATABASE_URL", raising=False)
+    args = MODULE.build_parser().parse_args(
+        [
+            "--tenant-id",
+            "default",
+            "--created-before",
+            "2026-07-14T00:00:00Z",
+            "--expected-host",
+            "migration-probe",
+            "--expected-database",
+            "migration_probe",
+        ]
+    )
+
+    with pytest.raises(MODULE.MaintenanceDatabaseTargetError) as exc_info:
+        MODULE.run(args)
+
+    assert exc_info.value.code == "database_url_required"
+
+
+def test_partial_failure_payload_discloses_committed_progress() -> None:
+    summary = {
+        "run_id": "historical-import-run-1",
+        "batches": 2,
+        "inspected": 4,
+        "created": 3,
+        "already_linked": 1,
+        "last_cursor": {
+            "created_at": "2026-07-10T00:00:00Z",
+            "claim_id": "claim-4",
+        },
+    }
+    details = MODULE._execution_progress(
+        summary,
+        current_batch_conflict_ids=["claim-5"],
+    )
+    error = MODULE.BackfillCommandError(
+        "current batch rolled back",
+        exit_code=MODULE.EXIT_CONFLICT,
+        code="legacy_claim_conflict",
+        details=details,
+    )
+
+    payload = MODULE._error_payload(error, code=error.code, details=error.details)
+
+    assert payload["details"] == {
+        "run_id": "historical-import-run-1",
+        "partial_commit": True,
+        "committed_batches": 2,
+        "committed_claims": 4,
+        "created": 3,
+        "already_linked": 1,
+        "last_committed_cursor": {
+            "created_at": "2026-07-10T00:00:00Z",
+            "claim_id": "claim-4",
+        },
+        "current_batch_conflict_ids": ["claim-5"],
+    }

server/tests/test_expense_case_legacy_backfill.py

@@ -0,0 +1,385 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from decimal import Decimal
+
+import pytest
+from sqlalchemy import create_engine, func, select
+from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.pool import StaticPool
+
+from app.db.base import Base
+from app.models.expense_case import BusinessEvent, ExpenseCase, ExpenseCaseLink
+from app.models.financial_record import ExpenseClaim
+from app.services.expense_case_legacy_backfill import (
+    HISTORICAL_CLAIM_IMPORTED_EVENT,
+    HISTORICAL_IMPORT_DELIVERY_STATUS,
+    ExpenseCaseLegacyBackfillService,
+    LegacyBackfillDisposition,
+)
+
+CUTOFF = datetime(2026, 7, 14, tzinfo=UTC)
+
+
+def build_session() -> Session:
+    engine = create_engine(
+        "sqlite+pysqlite:///:memory:",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
+    Base.metadata.create_all(bind=engine)
+    return sessionmaker(bind=engine, autoflush=False, autocommit=False)()
+
+
+def build_claim(claim_no: str, *, created_at: datetime | None = None) -> ExpenseClaim:
+    return ExpenseClaim(
+        claim_no=claim_no,
+        employee_name="张三",
+        department_name="市场部",
+        project_code="PRJ-BACKFILL",
+        expense_type="travel",
+        reason="历史客户拜访",
+        location="上海",
+        amount=Decimal("128.50"),
+        currency="CNY",
+        invoice_count=1,
+        occurred_at=datetime(2026, 7, 10, 9, 0, tzinfo=UTC),
+        submitted_at=datetime(2026, 7, 11, 10, 0, tzinfo=UTC),
+        status="submitted",
+        approval_stage="直属领导审批",
+        risk_flags_json=[],
+        created_at=created_at or datetime(2026, 7, 10, 8, 0, tzinfo=UTC),
+        updated_at=datetime(2026, 7, 11, 10, 0, tzinfo=UTC),
+    )
+
+
+def seed_link(db: Session, claim: ExpenseClaim, *, tenant_id: str = "default") -> None:
+    expense_case = ExpenseCase(
+        tenant_id=tenant_id,
+        case_no=f"CASE-{claim.claim_no}",
+        scene_code="travel",
+        title=claim.reason,
+        current_stage="reviewing",
+        status="active",
+    )
+    db.add(expense_case)
+    db.flush()
+    db.add(
+        ExpenseCaseLink(
+            tenant_id=tenant_id,
+            expense_case_id=expense_case.id,
+            resource_type="expense_claim",
+            resource_id=claim.id,
+            relation_type="claim",
+        )
+    )
+    db.flush()
+
+
+def seed_event_without_link(db: Session, claim: ExpenseClaim) -> None:
+    expense_case = ExpenseCase(
+        tenant_id="default",
+        case_no=f"CASE-ORPHAN-{claim.claim_no}",
+        scene_code="travel",
+        title=claim.reason,
+        current_stage="reviewing",
+        status="active",
+    )
+    db.add(expense_case)
+    db.flush()
+    db.add(
+        BusinessEvent(
+            tenant_id="default",
+            expense_case_id=expense_case.id,
+            aggregate_type="expense_claim",
+            aggregate_id=claim.id,
+            event_type="claim_submitted",
+            event_version=1,
+            idempotency_key=f"orphan:{claim.id}",
+            correlation_id="orphan-event",
+            actor_id="system",
+            actor_type="system",
+            payload_json={},
+            delivery_status="pending",
+            occurred_at=datetime(2026, 7, 11, tzinfo=UTC),
+        )
+    )
+    db.flush()
+
+
+def table_counts(db: Session) -> tuple[int, int, int]:
+    return (
+        db.scalar(select(func.count()).select_from(ExpenseCase)) or 0,
+        db.scalar(select(func.count()).select_from(ExpenseCaseLink)) or 0,
+        db.scalar(select(func.count()).select_from(BusinessEvent)) or 0,
+    )
+
+
+def test_requires_explicit_tenant_and_timezone_aware_cutoff() -> None:
+    with build_session() as db:
+        with pytest.raises(ValueError, match="tenant_id"):
+            ExpenseCaseLegacyBackfillService(db, tenant_id="", cutoff=CUTOFF)
+        with pytest.raises(ValueError, match="timezone"):
+            ExpenseCaseLegacyBackfillService(
+                db,
+                tenant_id="default",
+                cutoff=datetime(2026, 7, 14),
+            )
+
+
+def test_fingerprint_is_stable_and_idempotency_key_is_tenant_scoped() -> None:
+    with build_session() as db:
+        claim = build_claim("EXP-HISTORY-FINGERPRINT")
+        db.add(claim)
+        db.flush()
+        default_service = ExpenseCaseLegacyBackfillService(
+            db,
+            tenant_id="default",
+            cutoff=CUTOFF,
+        )
+        tenant_service = ExpenseCaseLegacyBackfillService(
+            db,
+            tenant_id="tenant-b",
+            cutoff=CUTOFF,
+        )
+
+        first_fingerprint = default_service.source_fingerprint(claim)
+        assert default_service.source_fingerprint(claim) == first_fingerprint
+        assert default_service.idempotency_key(claim.id) == (
+            f"historical-import:v1:default:{claim.id}"
+        )
+        assert tenant_service.idempotency_key(claim.id) == (
+            f"historical-import:v1:tenant-b:{claim.id}"
+        )
+
+        claim.status = "approved"
+        assert default_service.source_fingerprint(claim) != first_fingerprint
+
+
+def test_long_tenant_idempotency_key_is_stably_hashed_to_database_limit() -> None:
+    with build_session() as db:
+        service = ExpenseCaseLegacyBackfillService(
+            db,
+            tenant_id="t" * 64,
+            cutoff=CUTOFF,
+        )
+
+        first = service.idempotency_key("c" * 36)
+        second = service.idempotency_key("c" * 36)
+
+        assert first == second
+        assert first.startswith("historical-import:v1:sha256:")
+        assert len(first) <= 120
+
+
+def test_preview_is_read_only_and_classifies_linked_and_orphan_event_conflicts() -> None:
+    with build_session() as db:
+        eligible = build_claim("EXP-HISTORY-ELIGIBLE")
+        linked = build_claim("EXP-HISTORY-LINKED", created_at=CUTOFF - timedelta(days=3))
+        conflict = build_claim("EXP-HISTORY-CONFLICT", created_at=CUTOFF - timedelta(days=2))
+        future = build_claim("EXP-HISTORY-FUTURE", created_at=CUTOFF + timedelta(seconds=1))
+        db.add_all([eligible, linked, conflict, future])
+        db.flush()
+        seed_link(db, linked)
+        seed_event_without_link(db, conflict)
+        db.commit()
+
+        counts_before = table_counts(db)
+        preview = ExpenseCaseLegacyBackfillService(
+            db,
+            tenant_id="default",
+            cutoff=CUTOFF,
+        ).preview(batch_size=10)
+
+        assert preview.inspected == 3
+        assert preview.eligible == 1
+        assert preview.linked == 1
+        assert preview.conflicts == 1
+        assert not preview.has_more
+        assert table_counts(db) == counts_before
+        dispositions = {item.claim_no: item.disposition for item in preview.items}
+        assert dispositions == {
+            "EXP-HISTORY-ELIGIBLE": LegacyBackfillDisposition.ELIGIBLE,
+            "EXP-HISTORY-LINKED": LegacyBackfillDisposition.LINKED,
+            "EXP-HISTORY-CONFLICT": LegacyBackfillDisposition.CONFLICT,
+        }
+
+
+def test_apply_batch_writes_honest_snapshot_without_committing_and_is_idempotent() -> None:
+    with build_session() as db:
+        claim = build_claim("EXP-HISTORY-APPLY")
+        db.add(claim)
+        db.commit()
+        backfilled_at = datetime(2026, 7, 14, 12, 30, tzinfo=UTC)
+        service = ExpenseCaseLegacyBackfillService(
+            db,
+            tenant_id="default",
+            cutoff=CUTOFF,
+        )
+
+        result = service.apply_batch(
+            run_id="legacy-backfill-run-1",
+            batch_size=10,
+            backfilled_at=backfilled_at,
+        )
+
+        assert result.created == 1
+        assert result.skipped_linked == 0
+        assert result.conflicts == 0
+        event = db.scalar(select(BusinessEvent))
+        assert event is not None
+        assert event.event_type == HISTORICAL_CLAIM_IMPORTED_EVENT
+        assert event.idempotency_key == (f"historical-import:v1:default:{claim.id}")
+        assert event.idempotency_key == service.idempotency_key(claim.id)
+        assert event.actor_id == "system"
+        assert event.actor_type == "system"
+        assert event.delivery_status == HISTORICAL_IMPORT_DELIVERY_STATUS
+        assert event.occurred_at == backfilled_at
+        assert event.payload_json["history_reconstructed"] is False
+        assert event.payload_json["backfill_version"] == 1
+        assert event.payload_json["backfilled_at"] == "2026-07-14T12:30:00Z"
+        assert event.payload_json["performed_by"] == "expense_case_legacy_backfill_cli"
+        assert event.payload_json["source_times"] == {
+            "occurred_at": "2026-07-10T09:00:00Z",
+            "submitted_at": "2026-07-11T10:00:00Z",
+            "created_at": "2026-07-10T08:00:00Z",
+            "updated_at": "2026-07-11T10:00:00Z",
+        }
+        assert event.payload_json["source_fingerprint"].startswith("sha256:")
+        assert "未重建" in event.payload_json["reason"]
+
+        db.rollback()
+        assert table_counts(db) == (0, 0, 0)
+
+        first_apply = service.apply_batch(
+            run_id="legacy-backfill-run-2",
+            batch_size=10,
+            backfilled_at=backfilled_at,
+        )
+        db.commit()
+        repeated_apply = service.apply_batch(
+            run_id="legacy-backfill-run-3",
+            batch_size=10,
+            backfilled_at=backfilled_at + timedelta(minutes=1),
+        )
+
+        assert first_apply.created == 1
+        assert repeated_apply.created == 0
+        assert repeated_apply.skipped_linked == 1
+        assert table_counts(db) == (1, 1, 1)
+
+
+def test_apply_batch_reports_event_without_link_as_conflict() -> None:
+    with build_session() as db:
+        claim = build_claim("EXP-HISTORY-ORPHAN")
+        db.add(claim)
+        db.flush()
+        seed_event_without_link(db, claim)
+        db.commit()
+
+        result = ExpenseCaseLegacyBackfillService(
+            db,
+            tenant_id="default",
+            cutoff=CUTOFF,
+        ).apply_batch(
+            run_id="legacy-backfill-conflict",
+            batch_size=10,
+        )
+
+        assert result.created == 0
+        assert result.conflicts == 1
+        assert result.items[0].disposition is LegacyBackfillDisposition.CONFLICT
+        assert "without an expense case link" in result.items[0].reason
+        assert db.scalar(select(func.count()).select_from(ExpenseCaseLink)) == 0
+
+
+def test_preview_rejects_damaged_link_tenant_and_unlinked_case() -> None:
+    with build_session() as db:
+        damaged_link_claim = build_claim("EXP-HISTORY-DAMAGED-LINK")
+        unlinked_case_claim = build_claim(
+            "EXP-HISTORY-UNLINKED-CASE",
+            created_at=CUTOFF - timedelta(days=2),
+        )
+        db.add_all([damaged_link_claim, unlinked_case_claim])
+        db.flush()
+
+        foreign_case = ExpenseCase(
+            tenant_id="other-tenant",
+            case_no=f"CASE-{damaged_link_claim.claim_no}",
+            scene_code="travel",
+            title=damaged_link_claim.reason,
+            current_stage="reviewing",
+            status="active",
+        )
+        orphan_case = ExpenseCase(
+            tenant_id="default",
+            case_no=f"CASE-{unlinked_case_claim.claim_no}",
+            scene_code="travel",
+            title=unlinked_case_claim.reason,
+            current_stage="reviewing",
+            status="active",
+        )
+        db.add_all([foreign_case, orphan_case])
+        db.flush()
+        db.add(
+            ExpenseCaseLink(
+                tenant_id="default",
+                expense_case_id=foreign_case.id,
+                resource_type="expense_claim",
+                resource_id=damaged_link_claim.id,
+                relation_type="claim",
+            )
+        )
+        db.commit()
+
+        preview = ExpenseCaseLegacyBackfillService(
+            db,
+            tenant_id="default",
+            cutoff=CUTOFF,
+        ).preview(batch_size=10)
+
+        assert preview.conflicts == 2
+        reasons = {item.claim_no: item.reason for item in preview.items}
+        assert "invalid tenant" in reasons["EXP-HISTORY-DAMAGED-LINK"]
+        assert "without an expense claim link" in reasons["EXP-HISTORY-UNLINKED-CASE"]
+
+
+def test_preview_cursor_provides_stable_keyset_batches() -> None:
+    with build_session() as db:
+        first = build_claim("EXP-HISTORY-001", created_at=CUTOFF - timedelta(days=3))
+        second = build_claim("EXP-HISTORY-002", created_at=CUTOFF - timedelta(days=2))
+        db.add_all([first, second])
+        db.commit()
+        service = ExpenseCaseLegacyBackfillService(
+            db,
+            tenant_id="default",
+            cutoff=CUTOFF,
+        )
+
+        page_one = service.preview(batch_size=1)
+        page_two = service.preview(batch_size=1, after=page_one.next_cursor)
+
+        assert page_one.has_more
+        assert page_one.items[0].claim_no == "EXP-HISTORY-001"
+        assert not page_two.has_more
+        assert page_two.items[0].claim_no == "EXP-HISTORY-002"
+
+
+def test_standard_event_delivery_status_remains_pending_by_default() -> None:
+    with build_session() as db:
+        claim = build_claim("EXP-NORMAL-EVENT")
+        db.add(claim)
+        db.commit()
+
+        _expense_case, event = ExpenseCaseLegacyBackfillService(
+            db,
+            tenant_id="default",
+            cutoff=CUTOFF,
+        ).expense_cases.record_claim_event(
+            claim,
+            event_type="claim_submitted",
+            actor_id="owner@example.com",
+            tenant_id="default",
+        )
+
+        assert event.delivery_status == "pending"

server/tests/test_maintenance_database_target.py

@@ -0,0 +1,225 @@
+from __future__ import annotations
+
+import pytest
+
+from app.db.maintenance_database_target import (
+    MaintenanceDatabaseTargetError,
+    is_disposable_maintenance_target,
+    parse_maintenance_database_target,
+    validate_maintenance_database_target,
+)
+
+
+def assert_target_error(code: str, callback) -> None:
+    with pytest.raises(MaintenanceDatabaseTargetError) as exc_info:
+        callback()
+    assert exc_info.value.code == code
+
+
+def test_parse_target_supports_hostname_and_hides_password() -> None:
+    target = parse_maintenance_database_target(
+        "postgresql+psycopg://finance:super-secret@db.internal:5544/x_financial"
+        "?application_name=backfill",
+        expected_host="DB.INTERNAL.",
+        expected_database="x_financial",
+    )
+
+    assert target.host == "db.internal"
+    assert target.port == 5544
+    assert target.database == "x_financial"
+    assert target.username == "finance"
+    assert target.exact_target == "db.internal:5544/x_financial"
+    assert "super-secret" not in target.sanitized_url
+    assert "***" in target.sanitized_url
+    assert "application_name=backfill" in target.sanitized_url
+    assert target.is_disposable is False
+
+
+def test_parse_target_supports_ipv4_and_defaults_postgresql_port() -> None:
+    target = parse_maintenance_database_target(
+        "postgresql://finance:password@127.0.0.1/x_financial",
+        expected_host="127.0.0.1",
+        expected_database="x_financial",
+    )
+
+    assert target.port == 5432
+    assert target.exact_target == "127.0.0.1:5432/x_financial"
+    assert "password" not in target.sanitized_url
+
+
+def test_parse_target_redacts_sensitive_query_values() -> None:
+    target = parse_maintenance_database_target(
+        "postgresql://finance:authority-secret@db.internal/x_financial"
+        "?sslpassword=query-secret&application_name=safe-name",
+        expected_host="db.internal",
+        expected_database="x_financial",
+    )
+
+    assert "authority-secret" not in target.sanitized_url
+    assert "query-secret" not in target.sanitized_url
+    assert "application_name=safe-name" in target.sanitized_url
+
+
+@pytest.mark.parametrize(
+    "routing_query",
+    [
+        "host=x-financial-local-postgres",
+        "port=6432",
+        "dbname=production",
+        "user=other-user",
+        "service=production-service",
+        "options=-csearch_path%3Dother_schema",
+    ],
+)
+def test_parse_target_rejects_query_parameters_that_can_override_route(
+    routing_query: str,
+) -> None:
+    assert_target_error(
+        "database_routing_query_forbidden",
+        lambda: parse_maintenance_database_target(
+            f"postgresql://finance:password@migration-probe/migration_probe?{routing_query}",
+            expected_host="migration-probe",
+            expected_database="migration_probe",
+        ),
+    )
+
+
+@pytest.mark.parametrize(
+    ("database_url", "code"),
+    [
+        ("", "database_url_required"),
+        ("not a database url", "invalid_database_url"),
+        ("sqlite:///tmp/test.db", "postgresql_required"),
+        ("postgresql:///x_financial", "database_host_required"),
+        ("postgresql://finance@db.internal", "database_name_required"),
+    ],
+)
+def test_parse_target_rejects_invalid_or_incomplete_urls(database_url: str, code: str) -> None:
+    assert_target_error(
+        code,
+        lambda: parse_maintenance_database_target(
+            database_url,
+            expected_host="db.internal",
+            expected_database="x_financial",
+        ),
+    )
+
+
+@pytest.mark.parametrize(
+    ("expected_host", "expected_database", "code"),
+    [
+        ("", "x_financial", "expected_host_required"),
+        ("other-db", "x_financial", "expected_host_mismatch"),
+        ("db.internal", "", "expected_database_required"),
+        ("db.internal", "other_database", "expected_database_mismatch"),
+    ],
+)
+def test_parse_target_rejects_expected_target_mismatches(
+    expected_host: str,
+    expected_database: str,
+    code: str,
+) -> None:
+    assert_target_error(
+        code,
+        lambda: parse_maintenance_database_target(
+            "postgresql://finance:password@db.internal/x_financial",
+            expected_host=expected_host,
+            expected_database=expected_database,
+        ),
+    )
+
+
+@pytest.mark.parametrize(
+    ("host", "database", "expected"),
+    [
+        ("migration-probe", "migration_probe", True),
+        ("x-financial-migration-probe-123", "migration_probe_clone_123", True),
+        ("disposable-probe-7", "disposable_probe_7", True),
+        ("x-financial-local-postgres", "migration_probe", False),
+        ("migration-probe-7", "x_financial", False),
+    ],
+)
+def test_disposable_target_requires_markers_on_host_and_database(
+    host: str,
+    database: str,
+    expected: bool,
+) -> None:
+    assert is_disposable_maintenance_target(host=host, database=database) is expected
+
+
+def test_dry_run_allows_non_disposable_target_without_apply_override() -> None:
+    target = validate_maintenance_database_target(
+        "postgresql://finance:password@db.internal/x_financial",
+        expected_host="db.internal",
+        expected_database="x_financial",
+        apply=False,
+    )
+
+    assert target.is_disposable is False
+
+
+def test_apply_allows_disposable_target_without_non_disposable_override() -> None:
+    target = validate_maintenance_database_target(
+        "postgresql://finance:password@x-financial-migration-probe-1/migration_probe_1",
+        expected_host="x-financial-migration-probe-1",
+        expected_database="migration_probe_1",
+        apply=True,
+    )
+
+    assert target.is_disposable is True
+
+
+def test_apply_rejects_wrong_optional_confirmation_for_disposable_target() -> None:
+    assert_target_error(
+        "confirm_target_mismatch",
+        lambda: validate_maintenance_database_target(
+            "postgresql://finance:password@migration-probe/migration_probe",
+            expected_host="migration-probe",
+            expected_database="migration_probe",
+            apply=True,
+            confirm_target="other:5432/migration_probe",
+        ),
+    )
+
+
+def test_apply_rejects_non_disposable_target_without_explicit_override() -> None:
+    assert_target_error(
+        "non_disposable_apply_forbidden",
+        lambda: validate_maintenance_database_target(
+            "postgresql://finance:password@db.internal/x_financial",
+            expected_host="db.internal",
+            expected_database="x_financial",
+            apply=True,
+            confirm_target="db.internal:5432/x_financial",
+        ),
+    )
+
+
+@pytest.mark.parametrize("confirm_target", [None, "", "db.internal:5432/other_database"])
+def test_apply_rejects_non_disposable_target_without_exact_confirmation(
+    confirm_target: str | None,
+) -> None:
+    assert_target_error(
+        "confirm_target_mismatch",
+        lambda: validate_maintenance_database_target(
+            "postgresql://finance:password@db.internal/x_financial",
+            expected_host="db.internal",
+            expected_database="x_financial",
+            apply=True,
+            allow_non_disposable=True,
+            confirm_target=confirm_target,
+        ),
+    )
+
+
+def test_apply_allows_non_disposable_target_with_override_and_exact_confirmation() -> None:
+    target = validate_maintenance_database_target(
+        "postgresql://finance:password@db.internal/x_financial",
+        expected_host="db.internal",
+        expected_database="x_financial",
+        apply=True,
+        allow_non_disposable=True,
+        confirm_target="db.internal:5432/x_financial",
+    )
+
+    assert target.exact_target == "db.internal:5432/x_financial"

web/src/utils/expenseCaseTimeline.js

@@ -26,6 +26,7 @@ const BUSINESS_STATUS_LABELS = {
 const EVENT_PRESENTATION = {
   claim_draft_created: { label: '报销草稿已创建', icon: 'mdi mdi-file-plus-outline', tone: 'info' },
   claim_draft_updated: { label: '报销草稿已更新', icon: 'mdi mdi-file-edit-outline', tone: 'info' },
+  historical_claim_imported: { label: '历史单据已纳入统一费用事件', icon: 'mdi mdi-history', tone: 'info' },
   application_submitted: { label: '费用申请已提交', icon: 'mdi mdi-send-outline', tone: 'info' },
   claim_submitted: { label: '报销单已提交', icon: 'mdi mdi-send-outline', tone: 'info' },
   approval_stage_completed: { label: '审批节点已完成', icon: 'mdi mdi-account-check-outline', tone: 'success' },
@@ -83,7 +84,30 @@ function buildStateTransitionSummary(payload) {
   return transition || (nextStage ? `当前节点：${nextStage}` : '')
 }
 
+function buildHistoricalClaimImportSummary(payload) {
+  const currentStatus = resolveStatusLabel(readField(payload, 'next_status', 'nextStatus'))
+  const currentStage = compactText(readField(payload, 'next_approval_stage', 'nextApprovalStage'))
+  const currentStateParts = []
+
+  if (currentStatus) {
+    currentStateParts.push(`纳入时状态：${currentStatus}`)
+  }
+  if (currentStage) {
+    currentStateParts.push(`纳入时节点：${currentStage}`)
+  }
+
+  const currentStateSummary = currentStateParts.length
+    ? `${currentStateParts.join(' · ')}。`
+    : ''
+
+  return `系统已根据原单据建立费用事件档案。${currentStateSummary}导入前办理记录以原单据为准，后续动作将在此持续记录。`
+}
+
 function buildEventSummary(eventType, payload) {
+  if (eventType === 'historical_claim_imported') {
+    return buildHistoricalClaimImportSummary(payload)
+  }
+
   const reason = compactText(payload?.reason)
   if (reason) {
     return reason
@@ -159,7 +183,7 @@ function normalizeEvent(event, index) {
     icon: presentation.icon,
     tone: presentation.tone,
     summary: buildEventSummary(eventType, payload),
-    actorLabel: resolveActorLabel(event),
+    actorLabel: eventType === 'historical_claim_imported' ? '系统' : resolveActorLabel(event),
     occurredAt: compactText(occurredAt),
     occurredAtLabel: formatOccurredAt(occurredAt),
     sortTimestamp: Number.isNaN(timestamp) ? null : timestamp,

web/tests/expense-case-timeline.test.mjs

@@ -100,6 +100,64 @@ test('expense case timeline gives returned events an actionable reason', () => {
   assert.equal(timeline.events[0].summary, '住宿票据缺少入住人信息')
 })
 
+test('expense case timeline explains a historical import without pretending it is a business action', () => {
+  const timeline = buildExpenseCaseTimelineViewModel({
+    events: [
+      {
+        id: 'historical-import-1',
+        event_type: 'historical_claim_imported',
+        actor_id: 'migration-worker',
+        actor_type: 'user',
+        occurred_at: '2026-07-14T03:00:00Z',
+        correlation_id: 'private-correlation',
+        delivery_status: 'pending',
+        payload_json: {
+          reason: '不应覆盖历史导入事件的专用说明',
+          next_status: 'submitted',
+          next_approval_stage: '直属领导审批',
+          migration_batch_id: 'private-batch',
+          source_table: 'expense_claims'
+        }
+      }
+    ]
+  })
+
+  const [event] = timeline.events
+  assert.equal(event.label, '历史单据已纳入统一费用事件')
+  assert.equal(event.icon, 'mdi mdi-history')
+  assert.equal(event.tone, 'info')
+  assert.equal(event.actorLabel, '系统')
+  assert.equal(
+    event.summary,
+    '系统已根据原单据建立费用事件档案。纳入时状态：已提交 · 纳入时节点：直属领导审批。导入前办理记录以原单据为准，后续动作将在此持续记录。'
+  )
+  assert.doesNotMatch(
+    JSON.stringify(event),
+    /不应覆盖|private-correlation|pending|private-batch|expense_claims/
+  )
+})
+
+test('expense case timeline keeps the historical import summary honest when state is missing', () => {
+  const timeline = buildExpenseCaseTimelineViewModel({
+    events: [
+      {
+        id: 'historical-import-without-state',
+        event_type: 'historical_claim_imported',
+        actor_id: 'system',
+        occurred_at: '2026-07-14T03:00:00Z',
+        payload_json: {}
+      }
+    ]
+  })
+
+  const [event] = timeline.events
+  assert.equal(
+    event.summary,
+    '系统已根据原单据建立费用事件档案。导入前办理记录以原单据为准，后续动作将在此持续记录。'
+  )
+  assert.doesNotMatch(event.summary, /纳入时状态|纳入时节点|undefined|null/)
+})
+
 test('expense case timeline describes application archive against the paid reimbursement', () => {
   const timeline = buildExpenseCaseTimelineViewModel({
     events: [