feat(expenses): add transactional expense case events

document/development/2026-07-13/dev-logs/bugs/nested-audit-premature-commit.md

@@ -0,0 +1,8 @@
+## 修复记录
+
+- 11:57：记录 bug 修复：关联申请归档/解绑时嵌套审计提前提交业务事务。
+  - Git 提交检查：已执行 `git fetch --all --prune`、upstream 与 local-ahead 检查，未发现 upstream 新提交或本地 ahead 新提交；工作区原有规则表和历史开发文档改动已保留。
+  - 修改：`AuditLogService` 与 `AuditLogRepository` 增加由调用方控制的提交方式；`ExpenseClaimApplicationHandoffMixin` 的关联申请归档和解绑审计改为 `flush`，由付款/删除外层事务统一提交。
+  - 操作：新增费用事件事务回滚测试；在付款状态与结构化 Outbox 事件之间制造失败，确认付款、申请归档和嵌套审计能够整体回滚。
+  - 验证：容器 `x-financial-local-linux` 内新增及核心费用流程共 29 项测试通过，静态检查、OpenAPI 校验、启动脚本语法和 Alembic upgrade/downgrade 离线 SQL 验证通过。
+  - 影响：付款或关联解绑中途失败时，不再留下“申请已归档但付款/事件未提交”的半完成状态，为事务 Outbox 提供可靠边界。

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

@@ -0,0 +1,482 @@
+# AI 费用闭环与价值证明 概念文档
+
+更新时间：2026-07-13
+
+文档路径：document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/CONCEPT.md
+
+## 功能一句话
+
+把申请、消费、票据、报销、审批、付款、入账、分析和持续学习连接成同一个费用事件，让员工少填表、审批人只处理例外、财务能确认真实节省，并让 AI 在可控边界内越用越准确。
+
+## 背景与问题
+
+- 当前现状：项目已经具备费用申请、AI 建单、票据夹、OCR、预算检查、风险预审、动态审批、员工画像、财务看板、知识库、Agent trace、反馈表和 few-shot 样本等较完整的功能骨架。容器运行时 OpenAPI 已覆盖 156 个业务操作，功能广度已经足够。
+- 用户痛点：申请、票据、报销、审批和分析仍以多个页面、多个服务和多套状态存在。用户需要在不同入口之间理解系统，而不是由系统主动接住费用事件；移动端主流程仍有 mock；审批通过后的真实付款、回执、ERP 凭证和对账没有形成完整闭环。
+- AI 痛点：当前系统能够记录会话、运行轨迹、风险反馈和员工画像，但没有统一记录“AI 建议了什么、用户改了什么、后来是否退回、最终是否付款或产生节省”。AI 能看到历史，却不能稳定把业务结果转化为下一次的个性化、自动化和风险校准。
+- 企业痛点：现有看板主要回答花了多少、预算用了多少、发现了多少风险，尚不能可信回答省了多少钱、为什么省、哪些建议被执行、哪些费用仍可优化。
+- 工程问题：旧 `ReimbursementRequest` 与当前 `ExpenseClaim` 能力重叠，申请与报销又复用部分字段和 JSON 标记；审批、预算、付款、归档和关系引用部分混入 `risk_flags_json`，不利于事务一致性、事件追溯和持续演进。
+- 商业影响：如果不能持续证明处理效率、风险改善和真实节省，产品只能按“报销工具”定价；如果能形成价值闭环，则可以扩展为智能风控、预算经营、价值洞察和企业集成平台。
+- 为什么现在需要做：现有模块已经足够支撑一条完整费用闭环，下一阶段继续横向新增页面会扩大编排和数据割裂。应先收口费用领域、业务事件、AI 决策、学习记忆和节省归因，再逐步开放自动化。
+
+相关既有方案：
+
+- `document/development/2026-07-03/feature/ai-data-flywheel/CONCEPT.md`：作为本功能的 AI 样本回流、评测门禁、Prompt 版本和 Canary 子能力，不重复建设。
+- `document/development/AI意图规划器/UNIFIED_GATE_PIPELINE.md`：作为 AI 场景识别和后端唯一编排入口的架构前置，不在前端继续增加影子门控。
+
+## 目标与非目标
+
+### 目标
+
+- [G1] 建立统一 `Expense Case`，把一次费用从需求表达、申请、票据、报销、审批、付款、入账到归档视为同一业务事件。
+- [G2] 建立零录入报销体验，自动匹配申请、预算、票据、费用类型、项目、成本中心和历史偏好，让用户主要核对异常项。
+- [G3] 建立按动作授权的安全自动化等级，从解释、建议、预填、可逆自动化逐步升级到低风险直通。
+- [G4] 建立“AI 决策 → 用户反馈 → 工作流结果 → 分层记忆 → 再决策”的学习闭环。
+- [G5] 建立节省机会、执行任务、实际结果和财务确认组成的 Savings Ledger，区分风险暴露、预计节省和已实现节省。
+- [G6] 建立从描述性统计到异常归因、预算预测、供应商分析、政策模拟和节省任务的费用经营分析能力。
+- [G7] 建立企业租户、用量计量、模型成本和客户价值指标，为年度订阅、用量超额和增值模块提供可审计基础。
+- [G8] 保持高风险动作、资金支付、制度发布和敏感主数据变更的人为授权、审计和回滚能力。
+
+### 非目标
+
+- [NG1] 本轮不自建商旅、企业卡、银行或支付供应链网络，只定义标准连接器和业务事件契约。
+- [NG2] 本轮不一次性替换所有现有报销接口；先建立统一费用事件和旁路事件账本，再分阶段迁移旧模型。
+- [NG3] 本轮不做模型微调或无监督自训练，优先使用结构化反馈、few-shot、规则学习、Prompt 版本和回归门禁。
+- [NG4] 本轮不允许 AI 无人值守执行资金支付、高风险批准/驳回、制度发布和收款账户变更。
+- [NG5] 本轮不同时覆盖所有费用场景，首个试点只选择一个高频、可测量、可闭环的场景。
+- [NG6] 本轮不做未经授权的跨租户学习和企业间明细 Benchmark；后续只允许严格聚合、隐私保护并获得授权的对标能力。
+- [NG7] 本轮不把风险关联单据总额、暂缓付款金额或未采纳建议直接计为企业节省。
+
+## 用户与场景
+
+### 目标用户
+
+1. 报销人/申请人：少填字段、少整理票据、少补件、少追问进度。
+2. 审批人：只处理必要性、例外和高风险问题，快速获得证据与建议。
+3. 财务运营：减少重复审核、退回、对账和月末整理，集中处理异常。
+4. CFO/管理层：了解费用增长原因、预算趋势、节省机会和实际 ROI。
+5. 风控/审计：查看规则依据、风险证据、算法版本、人工覆盖和审计结果。
+6. 系统管理员/IT：配置租户、组织、权限、连接器、数据保留、模型成本和自动化上限。
+
+### 使用入口
+
+- AI 工作台自然语言入口。
+- 统一费用事件详情页。
+- 移动端拍票、报销和审批入口。
+- 票据夹和自动票据收件箱。
+- 审批例外工作台。
+- CFO 费用价值看板。
+- AI 记忆与自动化策略设置。
+
+### 核心场景
+
+1. 员工说“下周去上海出差三天”，系统自动生成合规申请、预算影响和可选节省方案，用户核对后提交。
+2. 消费期间，邮箱发票、拍照票据、企业卡或商旅订单进入统一票据收件箱，系统按时间、金额、地点和商户自动归集到费用事件。
+3. 行程结束后，系统自动生成报销草稿，只要求用户处理缺票、归属或异常金额。
+4. 提交前，系统按事实、规则、证据和风险等级给出绿色通过、黄色修正、红色复核结果。
+5. 审批人看到预算影响、历史相似单、政策依据、风险证据和建议意见，低风险单据进入快速通道。
+6. 财务完成付款、回执、ERP 凭证和对账，所有结果回写费用事件并形成审计链。
+7. 月末系统识别预算超支、供应商价格漂移、重复采购、异常路线和流程瓶颈，生成有负责人和目标金额的节省任务。
+8. 用户修正字段、审批人覆盖建议、财务确认误报或节省结果后，系统更新用户、部门和企业记忆，下次减少重复操作。
+
+### 异常场景
+
+- OCR、模型、向量检索或外部连接器不可用时，回退到人工录入或稳定规则，不阻塞草稿保存和主链路。
+- 票据、申请或费用事件匹配置信度不足时，只展示候选，不自动绑定。
+- 企业规则与个人偏好冲突时，企业规则优先，必须解释冲突原因。
+- 高风险、超金额阈值、敏感账户变更或证据不足时，强制人工确认。
+- 自动化动作失败时必须幂等、可重试、可撤销，并保留错误状态和操作证据。
+- 预计节省没有财务确认或实际结果时，只能标记为机会，不得计入客户 ROI。
+
+## 功能能力
+
+- [C1] 费用事件能力：统一表达申请、消费、票据、报销、审批、付款、入账和归档关系。
+- [C2] 零录入能力：自动抽取并匹配申请、预算、票据、费用类型、项目、成本中心、参与人和历史偏好。
+- [C3] 预审与修复能力：输出事实、规则、证据、判断和建议动作，并提供一键补件、修正和解释入口。
+- [C4] 审批例外能力：风险分级、证据摘要、预算影响、推荐路由、意见草稿、批量处理、委托、转交、加签和超时升级。
+- [C5] 支付入账能力：以连接器方式接收付款批次、支付回执、ERP 凭证、银行流水和对账结果。
+- [C6] 学习记忆能力：记录 AI 建议、用户修改、审批覆盖、退回、付款和审计结果，形成用户、部门、企业三级记忆。
+- [C7] 自动化策略能力：按动作、金额、场景、风险、置信度、证据、可逆性和抽检率配置自动化等级。
+- [C8] 节省价值能力：把节省机会、建议动作、负责人、目标时间、预计节省、实际节省和财务确认形成闭环。
+- [C9] 费用经营能力：费用结构、预算预测、异常归因、供应商分析、政策模拟、流程成本和客户 ROI。
+- [C10] 商业计量能力：企业租户、套餐、配额、用量、模型/OCR 成本、增值模块、客户贡献毛利和价值证明。
+- [C11] 状态与权限：服务端认证、租户隔离、角色权限、动作授权、审计日志和敏感数据保留策略。
+- [C12] 边界与降级：外部依赖失败时主链路可用；高风险动作始终保留人工控制；所有学习与自动化可关闭、遗忘或回滚。
+
+## 方案设计
+
+### 前端
+
+#### 统一费用事件
+
+- 新增或重构费用事件详情容器，不再让用户理解申请单、票据夹、报销草稿、审批单和付款状态之间的内部关系。
+- 页面按“计划 → 消费 → 报销 → 审批 → 付款/入账 → 复盘”展示时间线和当前待办。
+- 每个 AI 填充字段显示来源、置信度和修改入口；低置信度字段集中进入“需要确认”区。
+- 退回、补件和断点续办直接回到对应问题，不要求用户重新开始对话。
+
+#### 移动端
+
+- 把拍照、相册选票、报销列表、审批列表和 AI 助手接入真实后端。
+- 支持离线拍票、失败重试和后台上传状态。
+- 未实现能力必须隐藏或明确标为不可用，不保留无响应的主按钮。
+
+#### 审批例外工作台
+
+- 以风险、金额、预算影响、等待时长和证据完整度排序。
+- 支持批量处理低风险事项，行内保留真实按钮和键盘可访问入口。
+- 高风险审批展示模型版本、规则版本、政策依据、证据和人工覆盖原因。
+
+#### CFO 价值看板
+
+- 首页优先展示财务确认现金节省、已核验工时价值、安全智能直通率和重大风险护栏。
+- 支持按部门、项目、费用类型、供应商、城市、时间和费用事件下钻。
+- 每项节省可打开来源、基线、建议、执行、确认和去重证据。
+
+#### AI 记忆与自动化设置
+
+- 用户可以查看“系统记住了什么、为什么记住、在哪里使用”，并支持修改、忘记和关闭个性化。
+- 企业管理员配置部门/企业记忆边界、有效期、敏感等级和自动化上限。
+
+### 后端
+
+#### 费用领域与编排
+
+- 新增 `ExpenseCaseService` 作为跨阶段编排入口，避免继续扩大万能 `ExpenseClaimService`。
+- `ExpenseCaseService` 只负责阶段协调，申请、票据、报销、审批、支付、入账、记忆和节省由独立协作者负责。
+- 复用统一 AI 场景注册与 LangGraph 编排，前端只按后端返回的 plan/action 渲染，不再增加业务门控。
+
+#### 业务事件与 AI 决策
+
+- 所有关键动作写入 append-only `business_events`，使用 `correlation_id` 串联同一费用事件、Agent run、审批和外部连接器事件。
+- 业务状态变更与对应 Outbox 事件必须在同一数据库事务中持久化；消费者按事件 ID 幂等处理，避免审计、学习和 ROI 数据因异步失败永久丢失。
+- 只有画像刷新、分析聚合和消息通知等可重建派生任务允许异步失败并重试，不得把申请、提交、退回、审批、支付和入账事件降级为可丢弃旁路日志。
+- 每个 AI 输出写入 `ai_decisions`，记录建议值、置信度、证据、模型、Prompt、规则、政策版本、风险等级和自动化模式。
+- 用户接受、修改、拒绝或忽略写入 `ai_decision_feedback`；审批、退回、付款和审计写入 `workflow_outcomes`。
+
+#### 记忆与学习
+
+- `memory_entries` 支持用户、部门、企业作用域，以及 candidate/active/suppressed 状态。
+- `memory_evidence_links` 保留记忆与业务事件、决策、反馈、结果的来源关系。
+- 企业制度优先于部门基线，部门基线优先于个人偏好；冲突时返回解释。
+- 复用既有 AI 数据飞轮的 few-shot、golden case、Prompt 版本、Canary 和回归门禁。
+
+#### 节省与价值
+
+- `savings_opportunities` 记录基线、风险暴露、建议动作、预计节省、负责人和截止时间。
+- `savings_realizations` 记录执行结果、实际节省、财务确认人、证据、归因状态和去重键。
+- 风险关联金额、暂缓付款和未采纳建议不得自动转为实际节省。
+
+#### 连接器
+
+- 定义统一 `ExpenseConnector` 协议，覆盖票据邮箱、税务验真、企业卡、商旅、支付、银行流水、ERP、HR、SSO、企微/钉钉。
+- 外部事件必须具备幂等键、来源系统、原始事件 ID、发生时间、处理状态、错误码和重试次数。
+
+#### 商业计量
+
+- 以租户、套餐、模块和时间窗口记录模型、OCR、文档、分析任务、连接器与人工实施用量。
+- 成本事件与客户价值事件分开记账，支持计算客户贡献毛利、实施成本摊销和私有部署成本。
+- 套餐与配额只控制商业权益，不改变安全规则、租户隔离和高风险人工复核边界。
+
+### 算法与规则
+
+#### 自动化决策
+
+- 自动化等级按动作计算，不按 Agent 整体授权。
+- 动作风险、模型置信度、证据完整度、历史命中率、金额阈值、可逆性、企业上限和抽检率共同决定是否执行。
+- L0 只读解释；L1 建议；L2 预填；L3 可逆自动化；L4 低风险直通；L5 资金支付和高风险决策始终保留人工。
+
+#### 记忆激活
+
+- 明确偏好可由用户确认后立即激活。
+- 隐式偏好必须达到最小一致证据数，并通过时间衰减、敏感性、企业规则冲突和异常值检查。
+- 错误标注、违规习惯和一次性例外不能直接变为默认记忆。
+
+#### 风险与预审
+
+- 统一输出事实、规则、证据、风险、建议和可执行修复动作。
+- 使用已确认正/负样本校准误报与漏检；规则或 Prompt 发布前运行 golden case。
+- 新策略先进入 shadow，再 Canary，最后按动作开放自动化。
+
+#### 费用分析与节省
+
+- 描述性分析回答“发生了什么”；诊断分析回答“为什么”；建议分析回答“做什么”；价值闭环回答“是否执行并产生了多少结果”。
+- 供应商、费用类型、城市、项目和部门基线必须保存数据窗口、样本量、算法版本和政策版本。
+- 节省归因必须区分现金节省、可释放工时价值和不可货币化效率改善。
+
+### 数据与契约
+
+#### 核心数据对象
+
+- `expense_cases`：统一费用事件和当前阶段。
+- `expense_case_links`：连接申请、报销、票据、审批、付款、凭证和外部对象。
+- `business_events`：append-only 业务事件账本。
+- `ai_decisions`：结构化 AI 建议与版本证据。
+- `ai_decision_feedback`：接受、修改、拒绝和忽略。
+- `workflow_outcomes`：退回、补件、审批、付款、审计和最终结果。
+- `memory_entries` / `memory_evidence_links`：分层记忆和来源。
+- `automation_policies` / `automation_grants`：动作级自动化策略和授权。
+- `profile_baseline_snapshots`：持久化员工、部门、供应商、费用和流程基线。
+- `savings_opportunities` / `savings_realizations`：节省机会和实现结果。
+- `usage_meter_events`：租户、模块、模型、OCR、文档和分析用量。
+
+#### 最小事件词典
+
+- `expense_case_created`
+- `application_generated` / `application_submitted` / `application_approved`
+- `receipt_received` / `receipt_verified` / `ocr_corrected`
+- `field_suggested` / `field_accepted` / `field_edited` / `field_rejected`
+- `attachment_associated` / `draft_saved` / `claim_submitted`
+- `claim_returned` / `supplement_completed`
+- `risk_flagged` / `risk_confirmed` / `risk_false_positive`
+- `route_suggested` / `route_overridden`
+- `claim_approved` / `payment_requested` / `payment_completed`
+- `accounting_entry_created` / `reconciliation_completed`
+- `saving_opportunity_created` / `saving_action_completed` / `saving_confirmed`
+
+#### 状态枚举
+
+- Expense Case：`planning` / `approved_to_spend` / `spending` / `claiming` / `reviewing` / `paying` / `accounting` / `closed` / `cancelled`。
+- AI Decision：`suggested` / `accepted` / `edited` / `rejected` / `ignored` / `executed` / `rolled_back`。
+- Memory：`candidate` / `active` / `suppressed` / `expired` / `revoked`。
+- Automation：`explain` / `recommend` / `prefill` / `reversible_auto` / `low_risk_straight_through` / `human_only`。
+- Savings：`identified` / `accepted` / `in_progress` / `realized` / `verified` / `rejected` / `expired`。
+
+#### 兼容策略
+
+- 迁移初期可用 shadow 事件校验现有 `ExpenseClaim` 的映射；某个状态一旦正式纳入事件模型，其业务写入与 Outbox 事件必须进入同一事务。
+- 旧 `ReimbursementRequest` 进入只读兼容和迁移状态，停止新增第二套业务编排。
+- 现有 `risk_flags_json` 保持读取兼容，新审批、付款、归档和关系事件写入结构化表。
+- API 新字段优先追加，不在同一阶段破坏现有前端契约。
+- 所有表结构通过 Alembic 迁移，不继续在请求路径执行 DDL。
+
+#### 版本与审计
+
+- 事件记录 actor、tenant、source、run、model、Prompt、rule、policy 和 schema 版本。
+- 记忆、自动化策略、规则、Prompt 和基线均支持 supersede、有效期和回滚。
+- 高风险人工覆盖必须记录原因、证据和审批主体。
+
+### 权限与安全
+
+- 登录后签发服务端可验证会话或 JWT，服务端从会话和数据库解析用户、租户、角色和数据范围。
+- 移除客户端 `X-Auth-*` 作为授权事实来源；管理面、Bootstrap、设置和模型连通性接口必须受平台管理员保护。
+- P0 数据契约即引入最小 `tenant_id`、数据库约束、行级过滤、向量库命名空间和对象存储前缀隔离；删除传播、数据导出和私有部署加固可在商业化阶段继续完善。
+- 自动化权限按动作、金额、场景、风险和有效期授予，不使用全局“允许 Agent 自动执行”开关。
+- 收款账户变更、资金支付、制度发布、高风险驳回和敏感主数据变更执行双人或更高等级复核。
+
+### 降级策略
+
+- LLM 不可用：回退到规则和人工填写。
+- OCR 不可用：文件保留并进入待识别队列，用户可手工补录。
+- Qdrant/few-shot 不可用：使用 stable Prompt 和基础规则，不阻塞主流程。
+- 外部支付/ERP/税务连接器不可用：事件进入 retryable 状态，保留人工处理入口和幂等键。
+- 记忆冲突或可信度不足：只展示建议，不自动应用。
+- Savings 证据不足：保留机会状态，不进入已实现节省。
+
+## 算法与公式
+
+### 自动化资格分数
+
+```text
+automation_score
+= w1 * model_confidence
+ + w2 * evidence_completeness
+ + w3 * historical_precision
+ + w4 * reversibility
+ - w5 * action_risk
+ - w6 * amount_risk
+```
+
+变量说明：
+
+- `model_confidence`：模型或规则对当前建议的置信度。
+- `evidence_completeness`：发票、申请、预算、合同、订单和政策证据完整度。
+- `historical_precision`：相同租户、场景、动作和版本的历史正确率。
+- `reversibility`：动作是否可以无损撤销或回滚。
+- `action_risk`：动作类型风险，支付和制度发布最高。
+- `amount_risk`：金额相对企业阈值和历史基线的风险。
+- `w1...w6`：按企业和动作配置的权重。
+- 适用边界：分数只决定候选自动化等级，仍需满足企业硬性白名单、金额上限、抽检率和人审要求。
+
+### 记忆激活置信度
+
+```text
+memory_confidence
+= consistent_evidence_weight
+ + outcome_success_weight
+ + explicit_confirmation_weight
+ - conflict_weight
+ - age_decay
+```
+
+变量说明：
+
+- `consistent_evidence_weight`：多次一致修改或选择产生的证据。
+- `outcome_success_weight`：建议最终一次通过、付款或审计确认的权重。
+- `explicit_confirmation_weight`：用户或管理员明确确认的权重。
+- `conflict_weight`：与企业制度、部门规则或其他记忆冲突的惩罚。
+- `age_decay`：记忆随时间衰减。
+- 适用边界：敏感信息、一次性例外和违规行为不得依赖分数自动激活。
+
+### 安全智能直通率
+
+```text
+safe_straight_through_rate
+= qualified_completed_cases_without_manual_correction_or_return
+ / eligible_completed_cases
+```
+
+分子还必须满足必要审批完成，且事后抽样审计未发现重大问题。
+
+### 客户可验证 ROI
+
+```text
+verified_value
+= verified_cash_savings
+ + verified_releasable_labor_value
+
+customer_roi
+= (verified_value - customer_total_cost) / customer_total_cost
+```
+
+变量说明：
+
+- `verified_cash_savings`：客户财务确认的重复支付阻止、超标准调整、采购或预算优化等实际现金节省。
+- `verified_releasable_labor_value`：基于上线前后人工分钟、单据量和角色完全成本计算，并经客户认可的工时价值。
+- `customer_total_cost`：订阅、用量、实施、集成和客户内部运营成本。
+- 适用边界：现金节省和工时价值分开披露；风险暴露、未采纳建议和暂缓付款不得计入。
+
+### 客户贡献毛利
+
+```text
+customer_contribution_margin
+= subscription_revenue
+ + usage_revenue
+ + module_revenue
+ + verified_savings_share
+ - llm_ocr_storage_cost
+ - third_party_cost
+ - support_cost
+ - amortized_implementation_cost
+```
+
+用于判断高收入但高度定制或私有部署客户是否实际盈利。
+
+## 测试方案
+
+### 后端
+
+- Expense Case 状态机、关系绑定、幂等和非法状态跃迁单元测试。
+- Business Event、AI Decision、Feedback、Outcome、Memory、Automation Policy、Savings Ledger service 单元测试。
+- 服务端会话、租户隔离、角色和动作权限的正向/越权测试。
+- 连接器事件幂等、重试、回执、失败恢复和重复支付防护测试。
+- Alembic baseline、升级、回滚边界和旧数据迁移测试。
+- 现有报销、预算、风险、知识库和 Agent 回归测试。
+
+### 前端
+
+- 费用事件时间线、异常确认、断点续办和操作反馈视图模型测试。
+- 移动拍票、上传失败恢复、真实列表和审批 mutation 测试。
+- 审批例外工作台键盘操作、焦点管理和权限态测试。
+- CFO 价值看板对预计/实际/确认节省的展示和下钻测试。
+- AI 记忆查看、修改、忘记和关闭个性化测试。
+- 生产构建、lint、typecheck 和浏览器关键流程测试。
+
+### 算法与规则
+
+- 自动化分数、硬阈值、金额上限、动作白名单和抽检策略测试。
+- 记忆候选、激活、冲突、过期、撤销和污染样本测试。
+- 风险规则、Prompt、few-shot 和 golden case 回归门禁测试。
+- Savings 去重、基线、归因和确认状态测试。
+
+### 集成
+
+- 一句话申请 → 票据归集 → 自动报销 → 预审 → 审批 → 付款事件 → 入账 → 归档端到端。
+- AI 建议 → 用户修改 → 退回/通过 → 记忆候选 → 下次建议变化闭环。
+- 风险命中 → 人工确认/误报 → few-shot → 新版本回放 → Canary/回滚闭环。
+- 节省机会 → 负责人执行 → 实际结果 → 财务确认 → ROI 看板闭环。
+- 多租户同名员工、同号单据、向量检索和对象存储隔离测试。
+
+### 容器验证
+
+所有后端、集成、数据库迁移和外部依赖验证必须在项目容器内执行，单条测试命令最大超时 60s：
+
+```bash
+docker exec -w /app -e SERVER_VENV_DIR=/tmp/x-financial-server-venv \
+  local-x-financial-linux \
+  /tmp/x-financial-server-venv/bin/pytest -q server/tests/test_expense_case_service.py
+```
+
+如果本地 Compose 使用不同容器名，先通过 `docker ps` 确认当前主应用容器，不得改为宿主机直接运行后端测试。
+
+### 手工验证
+
+- 在真实容器页面完成申请、拍票、报销、退回补件、审批、付款和价值看板流程。
+- 检查每个 AI 建议是否能展示来源、置信度、修改结果和后续业务结果。
+- 检查用户能否查看、修改和忘记 AI 记忆。
+- 检查高风险动作无法绕过人工，低风险可逆动作能够撤销和回放。
+- 检查每项实际节省能够追溯到基线、建议、执行、确认人和证据。
+
+## 指标与验收
+
+以下目标在缺少真实客户基线前均为首轮试点的方向性目标，正式数值必须在试点基线采集后冻结。
+
+- [A1] 全流程验收：任一费用事件可以从申请追溯到票据、报销、审批、付款、凭证、归档、AI 决策和节省结果。
+- [A2] 体验指标：首个试点场景报销创建时间 P50 目标不超过 60 秒。
+- [A3] 自动化指标：可结构化字段的自动填充率方向性目标不低于 85%，且字段修改率持续下降。
+- [A4] 质量指标：首次提交完整率方向性目标不低于 90%，退回率和每单人工触点低于试点基线。
+- [A5] 智能指标：AI 字段采纳率、重复纠正率、建议接受率和记忆复用成功率可按租户、场景、版本统计。
+- [A6] 风险护栏：重大风险漏检率、误报干扰率、人工覆盖率和抽样审计结果可计算；自动化提升不得以护栏恶化为代价。
+- [A7] 价值指标：每项节省区分机会、预计、执行、实现和财务确认；客户月度 ROI 可回放。
+- [A8] 商业指标：早期可计算付费试点转年度合同率、持续产生可验证价值的客户率和按客户贡献毛利；成熟后计算 NRR。
+- [A9] 性能指标：同步页面接口 P95、AI 预填 P95、OCR 队列等待、后台任务恢复和连接器重试达到试点 SLA，具体阈值在基线后确定。
+- [A10] 安全指标：客户端无法伪造管理员、跨租户访问返回拒绝、敏感动作具备二次授权和审计。
+- [A11] 可观测性：每个费用事件、AI 决策、工具调用、连接器事件、自动化执行和失败重试使用统一 correlation ID。
+- [A12] 工程验收：相关后端定向测试、前端测试、lint、typecheck、构建和端到端验证在容器事实环境中通过。
+
+## 风险与开放问题
+
+### 风险
+
+- 范围过大：费用闭环、AI 学习、价值分析和商业化不能同时全量实现，需要按 P0/P1/P2 阶段交付。
+- 领域模型迁移：旧 `ReimbursementRequest`、`ExpenseClaim` 和 JSON 状态并存，必须旁路记录、小步迁移和双读校验。
+- 认证和租户：当前客户端身份头不适合自动化和 SaaS，多租户、记忆和高风险动作开发前必须修复。
+- 反馈投毒：一次点击或违规习惯不能直接成为记忆，需要候选态、最小样本、制度约束和结果权重。
+- 自动化失控：高准确率不代表高风险动作可以无人值守，必须按动作授权并支持 shadow、Canary、抽检和回滚。
+- 虚假节省：风险暴露金额、暂缓付款和工时估算容易被夸大，必须由客户财务确认并执行去重。
+- 外部依赖：税务、支付、银行、商旅、ERP 和消息平台连接器存在可用性、资质和交付周期风险。
+- 移动端成熟度：拍票基础已有，但主业务仍有 mock，必须先完成真实登录、列表、草稿和审批闭环。
+- 成本失控：高频 LLM、OCR、向量检索和私有部署可能降低毛利，需要按租户和模块计量成本。
+- 数据稀疏：本地开发库缺少报销、反馈、风险和 golden 样本，正式目标必须来自真实试点而不是演示数据。
+
+### 已处理依赖
+
+- 已有预算、票据夹、OCR、报销草稿、风险规则、审批路由、知识库、员工画像、Agent trace、few-shot 和财务看板可复用。
+- 已有 AI 数据飞轮概念与阶段 1 few-shot 实现，不重复建设样本检索底座。
+- 已有统一门控管道设计，可作为 AI 场景收口依据。
+
+### 待确认
+
+- 首个目标客户规模和部署形态：中小 SaaS、中大型企业 SaaS 或集团私有部署。
+- 首个 90 天付费试点场景：差旅报销、业务招待、日常费用或预算风控。
+- 是否进入企业支付、商旅预订或公司卡领域；如果不进入，应明确聚焦 AI 费控与经营分析。
+- 客户是否接受工时价值进入 ROI，以及对应角色完全成本口径。
+- L3/L4 自动化允许的动作、金额阈值、抽检率和授权主体。
+- 税务验真、银行、ERP、企微/钉钉等连接器的首批合作范围。
+- SaaS 定价的员工档位、包含用量、超额计价和增值模块边界。
+
+### 降级策略
+
+- 首期只选择一个费用场景做完整闭环，其他场景保留现有流程。
+- 正式切换前允许 shadow 事件用于一致性校验；正式切换后的关键业务状态与 Outbox 事件必须原子提交，失败时整体回滚并向用户返回可重试状态。
+- 画像刷新、分析聚合、消息通知等可重建派生任务失败时进入重试/死信队列，不阻塞已成功提交的关键业务状态。
+- 自动化默认停留在 L1/L2，只有试点评估和风险护栏通过后才逐项开放 L3/L4。
+- 外部连接器未完成前保留人工导入、确认和对账入口，但状态语义和审计事件按正式契约记录。
+
+## 本轮实现记录
+
+- 2026-07-13：完成产品能力、端到端费用旅程、AI 学习飞轮、KPI 和商业模式分析，形成总功能概念文档。
+- 2026-07-13（规划阶段）：只沉淀功能规划，没有修改业务代码、数据库结构或现有接口。
+- 2026-07-13（P0 基础切片）：新增 `ExpenseCase`、`ExpenseCaseLink`、`BusinessEvent` 模型与 `ExpenseCaseService`，提供 `/api/v1/expense-cases/by-claim/{claim_id}` 时间线查询接口。
+- 2026-07-13（P0 基础切片）：草稿、提交、退回、审批、申请转报销、付款和申请归档开始旁写结构化事件；事件具备租户字段、correlation、causation、幂等键和待投递状态，并与业务状态在同一事务提交。
+- 2026-07-13（事务修复）：关联申请归档/解绑的内部审计改为 `flush`，由外层业务事务统一提交，避免审计日志提前提交付款或归档状态。
+- 2026-07-13（迁移桥接）：新增第一条 migration-owned schema revision；服务启动先执行 Alembic，旧 `create_all` 明确排除三张迁移表。完整历史 schema baseline、正式数据库 upgrade/rollback 和停止旧 DDL 仍未完成。
+- 2026-07-13（验证）：容器内新增测试与既有差旅主链路回归共 24 项通过；Alembic PostgreSQL upgrade/downgrade 离线 SQL、启动脚本语法、OpenAPI 路由和新增文件静态检查通过。未对持久化开发数据库执行迁移。

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

@@ -0,0 +1,162 @@
+# AI 费用闭环与价值证明 开发 TODO
+
+更新时间：2026-07-13
+
+文档路径：document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/TODO.md
+
+## 使用规则
+
+- 每个 TODO 必须对应 `CONCEPT.md` 中的目标、能力、方案或验收点。
+- 只有完成实现并验证后，才能把 `[ ]` 改成 `[x]`。
+- 已完成项必须补充文件、接口、命令、容器测试或真实页面证据。
+- 如果需求发生变化，先更新 `CONCEPT.md`，再调整本 TODO。
+- 实施顺序遵循：P0 费用闭环与数据基础 → P1 安全自动化与学习 → P2 费用经营与价值证明 → P3 商业化复制。
+- 所有后端、集成、迁移和依赖外部服务的验证只允许在项目容器内运行，单条测试命令最大超时 60s。
+
+## 1. 调研与边界
+
+- [x] [CONCEPT: 背景与问题] 盘点申请、票据、报销、预算、风险、审批、付款、分析、Agent trace、反馈和 AI 数据飞轮现状。
+  证据：`server/src/app/api/v1/endpoints`、`server/src/app/services`、`web/src`、`mobile/app/src`、容器运行时 OpenAPI 共 156 个业务操作。
+- [x] [CONCEPT: 背景与问题] 确认当前主要断点是消费/取票、真实支付入账、结果回流、节省归因和商业计量，而不是页面数量不足。
+  证据：`CONCEPT.md`“背景与问题”；`expense_claim_approval_flow.py` 付款为状态更新；本地容器关键反馈/样本表聚合结果为 0。
+- [x] [CONCEPT: 目标与非目标] 确认本功能采用一个总功能点分阶段实施，不一次性替换所有现有接口，不自建支付和商旅供应链网络。
+  证据：`CONCEPT.md`“目标与非目标”。
+- [x] [CONCEPT: 风险与开放问题] 记录认证、租户、旧模型迁移、反馈污染、虚假节省、自动化失控、连接器和数据稀疏风险。
+  证据：`CONCEPT.md`“风险与开放问题”。
+## 2. 开发前置门禁：试点与基线
+
+- [ ] [CONCEPT: 待确认] 确认首个目标客户规模、部署形态、购买决策人和最小租户边界。
+- [ ] [CONCEPT: 待确认] 确认首个 90 天试点费用场景，只允许一个场景进入 P0 开发。
+- [ ] [CONCEPT: 待确认] 确认是否进入企业支付、商旅预订或公司卡领域，并冻结首期连接器范围。
+- [ ] [CONCEPT: 待确认] 确认客户 ROI 中现金节省与工时价值的口径、基线窗口和签字人。
+- [ ] [CONCEPT: 指标与验收] 在 P0 开发前采集人工分钟、退回率、完成周期、人工触点、风险反馈、费用金额和付款结果基线。
+- [ ] [CONCEPT: 指标与验收] 为 P0/P1/P2 分别冻结一条可纵向跑通的闭环验收，不以“模块代码已完成”代替业务结果。
+
+## 3. 契约与设计
+
+- [ ] [CONCEPT: 费用领域与编排] 定义 `Expense Case`、申请、票据、报销、审批、付款、凭证和归档的领域边界与迁移关系。
+- [ ] [CONCEPT: 数据与契约] 定义 `expense_cases`、`expense_case_links` 和最小状态机，明确非法状态跃迁。
+- [ ] [CONCEPT: 数据与契约] 定义 `business_events` 事件信封、事件词典、correlation ID、幂等键和版本策略。
+- [ ] [CONCEPT: 业务事件与 AI 决策] 定义 `ai_decisions`、`ai_decision_feedback` 和 `workflow_outcomes` 契约。
+- [ ] [CONCEPT: 记忆与学习] 定义 `memory_entries`、证据链接、优先级、有效期、敏感等级、撤销和遗忘契约。
+- [ ] [CONCEPT: 自动化决策] 定义动作风险、金额阈值、置信度、证据完整度、可逆性、抽检率和企业授权策略。
+- [ ] [CONCEPT: 节省与价值] 定义 Savings Ledger 的机会、执行、实现、确认、去重和归因状态。
+- [ ] [CONCEPT: 连接器] 定义票据邮箱、税务、企业卡、商旅、支付、银行、ERP、消息和 SSO 连接器协议。
+- [ ] [CONCEPT: 权限与安全] 定义服务端会话、租户数据范围、角色和动作级授权契约。
+- [ ] [CONCEPT: 指标与验收] 完成首个试点指标字典、基线采集方案、分子分母、数据源和负责人。
+- [ ] [CONCEPT: 方案设计] 完成分阶段架构评审，确认新增 service 不继续堆入 `ExpenseClaimService`、`UserAgentService` 或大型前端 composable。
+
+## 4. P0 后端实现：费用闭环与数据基础
+
+- [ ] [CONCEPT: 权限与安全] 实现服务端可验证会话/JWT，移除客户端身份头作为授权事实来源。
+- [ ] [CONCEPT: 权限与安全] 为管理面、Bootstrap、Settings、模型连通性、日志和规则接口补齐平台管理员保护。
+- [ ] [CONCEPT: 权限与安全] 为所有新增表和共享核心数据补齐最小 `tenant_id`、数据库约束、查询守卫及默认租户迁移。
+- [ ] [CONCEPT: 权限与安全] 为 Qdrant collection/namespace、对象存储前缀和缓存键补齐租户隔离回归测试。
+- [ ] [CONCEPT: 数据与契约] 建立 Alembic baseline 和正式迁移链，停止请求路径运行 DDL。
+- [x] [CONCEPT: 费用领域与编排] 新增 `ExpenseCaseService` 和费用事件查询接口，保持编排与具体职责分离。
+  证据：`server/src/app/services/expense_cases.py`、`server/src/app/api/v1/endpoints/expense_cases.py`、`GET /api/v1/expense-cases/by-claim/{claim_id}`；容器 OpenAPI 校验通过。
+- [ ] [CONCEPT: 数据与契约] 新增 `expense_cases`、`expense_case_links` 和 `business_events` 表及迁移。
+- [ ] [CONCEPT: 业务事件与 AI 决策] 新增 `ai_decisions`、`ai_decision_feedback`、`workflow_outcomes` 表及迁移。
+- [ ] [CONCEPT: 数据与契约] 为申请、票据、草稿、提交、退回、审批、付款和归档接入统一 correlation ID。
+- [ ] [CONCEPT: 业务事件与 AI 决策] 建立事务 Outbox：申请、提交、退回、审批、支付和入账状态与事件同事务提交，消费端按事件 ID 幂等处理。
+- [x] [CONCEPT: 业务事件与 AI 决策] 完成首批草稿、提交、退回、审批、申请转报销、付款和申请归档事件旁写，具备 correlation、causation、幂等键及事务回滚。
+  证据：`expense_claim_draft_flow.py`、`expense_claims.py`、`expense_claim_approval_flow.py`、`test_expense_case_service.py`；容器测试覆盖同 Case 关联、重复事件去重及 Outbox 失败整体回滚。
+- [x] [CONCEPT: 兼容策略] 建立迁移桥接：服务启动先执行 Alembic，旧 metadata bootstrap 排除 migration-owned 表。
+  证据：`server_start.sh`、`agent_foundation.py`、`20260713_0001_expense_case_business_events.py`；容器 `sh -n`、离线 upgrade/downgrade SQL 和 legacy bootstrap 排除测试通过。
+- [ ] [CONCEPT: 兼容策略] 正式切换前以 shadow 事件校验现有 `ExpenseClaim` 映射；切换后禁止关键事件可丢弃写入。
+- [ ] [CONCEPT: 兼容策略] 制定旧 `ReimbursementRequest` 只读兼容、迁移和停止新增编排的计划。
+- [ ] [CONCEPT: 兼容策略] 把新增审批、付款、归档和关系事件移出 `risk_flags_json`，保留旧数据读取兼容。
+- [ ] [CONCEPT: 数据与契约] 补齐撤回、取消、驳回、作废、补件、支付失败、对账异常和归档状态。
+- [ ] [CONCEPT: 连接器] 实现统一连接器基类、幂等、重试、错误状态和回执事件。
+- [ ] [CONCEPT: 连接器] 实现支付批次、回执、重复付款防护、ERP 凭证和对账的内部契约，首期允许 mock connector 但不得再只写单一“已付款”状态。
+- [ ] [CONCEPT: 降级策略] 将附件关联和关联报销草稿后台任务迁为可持久化、可恢复、可幂等的任务状态。
+
+## 5. P0 前端实现：一键报销与真实移动端
+
+- [ ] [CONCEPT: 统一费用事件] 新增统一费用事件时间线，串联计划、消费、报销、审批、付款/入账和复盘。
+- [ ] [CONCEPT: 统一费用事件] 自动匹配申请、预算、票据、费用类型、项目、成本中心和常用字段。
+- [ ] [CONCEPT: 统一费用事件] 增加“需要确认”区，只展示低置信、缺失或冲突字段。
+- [ ] [CONCEPT: 风险与预审] 风险卡提供一键补件、修正、解释和重新预审入口。
+- [ ] [CONCEPT: 统一费用事件] 退回后直接定位问题字段，支持断点续办，不要求重新发起对话。
+- [ ] [CONCEPT: 移动端] 接通真实登录恢复、路由守卫、报销列表、详情、草稿、上传和审批 API。
+- [ ] [CONCEPT: 移动端] 完成拍照/相册 → OCR → 票据夹 → 费用事件 → 报销草稿真实闭环。
+- [ ] [CONCEPT: 移动端] 删除或禁用没有真实行为的报销、审批、发送和语音主按钮。
+- [ ] [CONCEPT: 前端] 修复表格整行点击键盘不可达、弹窗焦点管理、焦点环和移动触控目标。
+- [ ] [CONCEPT: 前端] 移除页面级全局 `transform: scale()`，使用真实响应式布局承载费用闭环页面。
+
+## 6. P1 算法与规则实现：安全自动化与持续学习
+
+- [ ] [CONCEPT: 自动化决策] 实现 L0-L5 动作级自动化等级和资格计算器。
+- [ ] [CONCEPT: 自动化决策] 为每个动作实现硬白名单、金额上限、证据要求、抽检率和企业上限。
+- [ ] [CONCEPT: 风险与预审] 统一风险输出为事实、规则、证据、判断、建议动作和降级原因。
+- [ ] [CONCEPT: 记忆激活] 实现 candidate/active/suppressed/expired/revoked 记忆状态机。
+- [ ] [CONCEPT: 记忆激活] 实现用户、部门、企业记忆优先级、冲突解释、时间衰减和最小样本要求。
+- [ ] [CONCEPT: 记忆与学习] 从字段接受/修改/拒绝、退回、审批覆盖、付款和审计结果生成记忆证据。
+- [ ] [CONCEPT: 记忆与学习] 将已确认 few-shot 扩展到报销预审和审批辅助，并按租户、场景、制度版本过滤。
+- [ ] [CONCEPT: 风险与预审] 完成 golden case、Prompt/规则版本、Canary、回归门禁和自动回滚。
+- [ ] [CONCEPT: 自动化决策] 先上线 shadow，再按动作逐项开放 L3；L4 必须单独评审。
+- [ ] [CONCEPT: 降级策略] 对模型、OCR、Qdrant、连接器和记忆服务实现稳定降级和可观测状态。
+
+## 7. P1 前端实现：审批例外与 AI 记忆
+
+- [ ] [CONCEPT: 审批例外工作台] 按风险、金额、预算影响、等待时长和证据完整度排序审批事项。
+- [ ] [CONCEPT: 审批例外工作台] 展示必要性、预算、政策、相似单、风险证据和 AI 意见草稿。
+- [ ] [CONCEPT: 审批例外工作台] 补齐批量审批、委托、转交、加签、会签和超时升级交互。
+- [ ] [CONCEPT: 审批例外工作台] 接入真实企微/钉钉/邮件触达和处理结果回写。
+- [ ] [CONCEPT: AI 记忆与自动化设置] 新增“我的 AI 记忆”，支持来源解释、修改、忘记和关闭个性化。
+- [ ] [CONCEPT: AI 记忆与自动化设置] 新增企业记忆、保留策略、敏感等级和自动化上限管理。
+- [ ] [CONCEPT: 前端] 展示自动化动作、执行依据、撤销入口、抽检状态和版本信息。
+
+## 8. P2 实现：费用经营与价值证明
+
+- [ ] [CONCEPT: 节省与价值] 新增 `savings_opportunities`、`savings_realizations` 和去重/确认表结构。
+- [ ] [CONCEPT: 节省与价值] 实现风险暴露、预计节省、执行中、实际节省和财务确认的严格状态转换。
+- [ ] [CONCEPT: 费用分析与节省] 持久化员工、部门、费用类型、供应商、城市、项目和流程基线，记录窗口和样本量。
+- [ ] [CONCEPT: 费用分析与节省] 实现预算预测、异常归因、供应商价格漂移、重复小额浪费和政策模拟。
+- [ ] [CONCEPT: CFO 价值看板] 展示现金节省、工时价值、直通率、风险护栏、节省来源和责任人。
+- [ ] [CONCEPT: CFO 价值看板] 实现部门、项目、费用类型、供应商、城市、时间和单据下钻。
+- [ ] [CONCEPT: CFO 价值看板] 每项节省支持查看基线、建议、执行、实际结果、确认人和证据。
+- [ ] [CONCEPT: 指标与验收] 生成客户月度 ROI 报告，现金节省与工时价值分开披露。
+- [ ] [CONCEPT: 风险与开放问题] 建立节省归因复核、重复收益去重和客户财务签字流程。
+
+## 9. P3 实现：商业化与规模复制
+
+- [ ] [CONCEPT: 商业计量] 在 P0 最小租户隔离基础上新增套餐、配额、用量和增值模块授权模型。
+- [ ] [CONCEPT: 商业计量] 按客户、模型、OCR、文档、分析任务和模块记录成本。
+- [ ] [CONCEPT: 商业计量] 建立客户贡献毛利、实施成本摊销和私有部署成本看板。
+- [ ] [CONCEPT: 目标与非目标] 定义年度基础订阅、用量超额、智能风控、预算经营、价值洞察、企业集成和私有部署包。
+- [ ] [CONCEPT: 目标与非目标] 仅对财务确认的已实现节省提供可选节省分成合同。
+- [ ] [CONCEPT: 连接器] 建立 ERP、HR、SSO、支付、电子档案和消息平台标准实施模板。
+- [ ] [CONCEPT: 权限与安全] 完成删除传播、数据导出、审计增强和私有部署安全验收，不把基础租户隔离留到本阶段。
+
+## 10. 测试与验证
+
+- [x] [CONCEPT: 测试方案] 完成 Expense Case/Link、业务事件幂等、租户边界、同事务回滚、申请转报销同 Case 和付款归档事件首批测试。
+  证据：容器内 `pytest -q server/tests/test_expense_case_service.py` 7 项通过；联合差旅主链路定向回归共 24 项通过。
+- [ ] [CONCEPT: 测试方案] 为 Expense Case 状态机、事件账本、AI 决策、结果、记忆、自动化和节省服务补充单元测试。
+- [ ] [CONCEPT: 测试方案] 为服务端会话、管理员保护、租户隔离、跨租户访问和敏感动作补充安全回归测试。
+- [ ] [CONCEPT: 测试方案] 为 Alembic baseline、升级、旧数据迁移和回滚边界补充 Postgres 集成测试。
+- [ ] [CONCEPT: 测试方案] 为连接器幂等、重试、回执、失败恢复、重复付款和对账补充测试。
+- [ ] [CONCEPT: 测试方案] 跑通申请 → 票据 → 报销 → 预审 → 审批 → 付款 → 入账 → 归档端到端。
+- [ ] [CONCEPT: 测试方案] 跑通 AI 建议 → 用户修改 → 工作流结果 → 记忆激活 → 下次建议变化闭环。
+- [ ] [CONCEPT: 测试方案] 跑通风险反馈 → few-shot → golden case → Canary → 回滚闭环。
+- [ ] [CONCEPT: 测试方案] 跑通节省机会 → 执行 → 实现 → 财务确认 → ROI 看板闭环。
+- [ ] [CONCEPT: 测试方案] 补充前端组件、视图模型、键盘操作、响应式、移动真实接口和构建验证。
+- [ ] [CONCEPT: 测试方案] 所有后端、集成和迁移测试在当前主应用容器内执行，单条命令最大超时 60s。
+- [ ] [CONCEPT: 指标与验收] 记录测试、lint、typecheck、构建、端到端和未覆盖风险证据。
+
+## 11. 分阶段试点与价值验证
+
+- [ ] [CONCEPT: 指标与验收] 确认开发前置门禁已完成，试点场景、客户群体、基线和纵向验收未发生未经评审的漂移。
+- [ ] [CONCEPT: 自动化决策] 前四周只运行建议和 shadow，不开放高风险自动化。
+- [ ] [CONCEPT: 指标与验收] 验证报销创建时间、自动填充率、首次提交完整率、人工触点和风险护栏。
+- [ ] [CONCEPT: 节省与价值] 由客户财务确认现金节省、工时价值、归因口径和去重结果。
+- [ ] [CONCEPT: 商业计量] 计算试点客户的订阅、模型、OCR、实施、支持和集成贡献毛利。
+- [ ] [CONCEPT: 指标与验收] 输出 90 天 ROI 报告和年度合同/扩展模块建议。
+
+## 12. 文档收尾
+
+- [ ] [CONCEPT: 指标与验收] 将试点实际基线替换方向性目标，冻结正式验收阈值。
+- [ ] [CONCEPT: 风险与开放问题] 更新目标客户、试点场景、支付边界、连接器范围和自动化授权结论。
+- [ ] [CONCEPT: 本轮实现记录] 每个阶段完成后补充实现文件、迁移、接口、测试和真实页面证据。
+- [ ] [CONCEPT: 功能一句话] 确认最终实现持续服务于“不用填表、少被退回、真正省钱”的核心结果。

server/alembic/versions/20260713_0001_expense_case_business_events.py

@@ -0,0 +1,162 @@
+"""add expense case and transactional business event tables
+
+This is the first migration-owned schema slice in a legacy database that still
+bootstraps older tables through SQLAlchemy metadata. The server startup runs
+this revision before the legacy bootstrap, and the legacy bootstrap explicitly
+excludes these migration-owned tables.
+
+Revision ID: 20260713_0001
+Revises:
+Create Date: 2026-07-13 11:45:00
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "20260713_0001"
+down_revision: str | None = None
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "expense_cases",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("tenant_id", sa.String(length=64), nullable=False),
+        sa.Column("case_no", sa.String(length=80), nullable=False),
+        sa.Column("scene_code", sa.String(length=50), nullable=False),
+        sa.Column("title", sa.String(length=200), nullable=False),
+        sa.Column("owner_employee_id", sa.String(length=36), nullable=True),
+        sa.Column("current_stage", sa.String(length=40), nullable=False),
+        sa.Column("status", sa.String(length=30), nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.func.now(),
+            nullable=False,
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.func.now(),
+            nullable=False,
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("tenant_id", "case_no", name="uq_expense_cases_tenant_case_no"),
+    )
+    op.create_index("ix_expense_cases_tenant_id", "expense_cases", ["tenant_id"])
+    op.create_index("ix_expense_cases_owner_employee_id", "expense_cases", ["owner_employee_id"])
+    op.create_index(
+        "ix_expense_cases_tenant_stage",
+        "expense_cases",
+        ["tenant_id", "current_stage"],
+    )
+    op.create_index(
+        "ix_expense_cases_tenant_status",
+        "expense_cases",
+        ["tenant_id", "status"],
+    )
+
+    op.create_table(
+        "expense_case_links",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("tenant_id", sa.String(length=64), nullable=False),
+        sa.Column("expense_case_id", sa.String(length=36), nullable=False),
+        sa.Column("resource_type", sa.String(length=50), nullable=False),
+        sa.Column("resource_id", sa.String(length=100), nullable=False),
+        sa.Column("relation_type", sa.String(length=50), nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.func.now(),
+            nullable=False,
+        ),
+        sa.ForeignKeyConstraint(["expense_case_id"], ["expense_cases.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("resource_type", "resource_id", name="uq_expense_case_links_resource"),
+    )
+    op.create_index("ix_expense_case_links_tenant_id", "expense_case_links", ["tenant_id"])
+    op.create_index(
+        "ix_expense_case_links_tenant_case",
+        "expense_case_links",
+        ["tenant_id", "expense_case_id"],
+    )
+
+    op.create_table(
+        "business_events",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("tenant_id", sa.String(length=64), nullable=False),
+        sa.Column("expense_case_id", sa.String(length=36), nullable=False),
+        sa.Column("aggregate_type", sa.String(length=50), nullable=False),
+        sa.Column("aggregate_id", sa.String(length=100), nullable=False),
+        sa.Column("event_type", sa.String(length=80), nullable=False),
+        sa.Column("event_version", sa.Integer(), nullable=False),
+        sa.Column("idempotency_key", sa.String(length=120), nullable=False),
+        sa.Column("correlation_id", sa.String(length=64), nullable=False),
+        sa.Column("causation_id", sa.String(length=64), nullable=True),
+        sa.Column("actor_id", sa.String(length=120), nullable=False),
+        sa.Column("actor_type", sa.String(length=30), nullable=False),
+        sa.Column("payload_json", sa.JSON(), nullable=False),
+        sa.Column("delivery_status", sa.String(length=20), nullable=False),
+        sa.Column("delivery_attempts", sa.Integer(), nullable=False),
+        sa.Column("last_delivery_error", sa.Text(), nullable=True),
+        sa.Column(
+            "occurred_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.func.now(),
+            nullable=False,
+        ),
+        sa.Column("published_at", sa.DateTime(timezone=True), nullable=True),
+        sa.ForeignKeyConstraint(["expense_case_id"], ["expense_cases.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "tenant_id",
+            "aggregate_type",
+            "aggregate_id",
+            "event_type",
+            "idempotency_key",
+            name="uq_business_event_idempotency",
+        ),
+    )
+    op.create_index("ix_business_events_tenant_id", "business_events", ["tenant_id"])
+    op.create_index("ix_business_events_event_type", "business_events", ["event_type"])
+    op.create_index("ix_business_events_correlation_id", "business_events", ["correlation_id"])
+    op.create_index(
+        "ix_business_events_tenant_case_time",
+        "business_events",
+        ["tenant_id", "expense_case_id", "occurred_at"],
+    )
+    op.create_index(
+        "ix_business_events_outbox",
+        "business_events",
+        ["delivery_status", "occurred_at"],
+    )
+    op.create_index(
+        "ix_business_events_aggregate",
+        "business_events",
+        ["aggregate_type", "aggregate_id"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_business_events_aggregate", table_name="business_events")
+    op.drop_index("ix_business_events_outbox", table_name="business_events")
+    op.drop_index("ix_business_events_tenant_case_time", table_name="business_events")
+    op.drop_index("ix_business_events_correlation_id", table_name="business_events")
+    op.drop_index("ix_business_events_event_type", table_name="business_events")
+    op.drop_index("ix_business_events_tenant_id", table_name="business_events")
+    op.drop_table("business_events")
+
+    op.drop_index("ix_expense_case_links_tenant_case", table_name="expense_case_links")
+    op.drop_index("ix_expense_case_links_tenant_id", table_name="expense_case_links")
+    op.drop_table("expense_case_links")
+
+    op.drop_index("ix_expense_cases_tenant_status", table_name="expense_cases")
+    op.drop_index("ix_expense_cases_tenant_stage", table_name="expense_cases")
+    op.drop_index("ix_expense_cases_owner_employee_id", table_name="expense_cases")
+    op.drop_index("ix_expense_cases_tenant_id", table_name="expense_cases")
+    op.drop_table("expense_cases")

server/server_start.sh

@@ -378,6 +378,12 @@ ensure_dependencies() {
     info "Server dependencies are ready."
 }
 
+run_database_migrations() {
+    info "Applying database migrations..."
+    "$PYTHON_BIN" -m alembic -c "$SCRIPT_DIR/alembic.ini" upgrade head
+    info "Database migrations are up to date."
+}
+
 start_server() {
     info "Starting FastAPI server..."
     info "Access: http://$SERVER_HOST:$SERVER_PORT"
@@ -402,6 +408,7 @@ case "$MODE" in
         ;;
     start)
         ensure_dependencies
+        run_database_migrations
         start_server
         ;;
     *)

server/src/app/api/deps.py

@@ -26,6 +26,7 @@ class CurrentUserContext:
     name: str
     role_codes: list[str]
     is_admin: bool
+    tenant_id: str = "default"
     department_name: str = ""
     cost_center: str = ""
     position: str = ""
@@ -101,6 +102,7 @@ def get_current_user(
         name=name or username,
         role_codes=role_codes,
         is_admin=is_admin,
+        tenant_id="default",
         department_name=(x_auth_department or "").strip(),
         cost_center=(x_auth_cost_center or "").strip(),
         position=(x_auth_position or "").strip(),

server/src/app/api/v1/endpoints/expense_cases.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.orm import Session
+
+from app.api.deps import CurrentUserContext, get_current_user, get_db
+from app.schemas.expense_case import ExpenseCaseTimelineRead
+from app.services.expense_cases import ExpenseCaseService
+from app.services.expense_claims import ExpenseClaimService
+
+router = APIRouter(prefix="/expense-cases")
+DbSession = Annotated[Session, Depends(get_db)]
+CurrentUser = Annotated[CurrentUserContext, Depends(get_current_user)]
+
+
+@router.get(
+    "/by-claim/{claim_id}",
+    response_model=ExpenseCaseTimelineRead,
+    summary="查询报销单所属费用事件时间线",
+    description="返回当前用户有权查看的费用事件、关联单据和结构化业务事件。",
+)
+def get_expense_case_by_claim(
+    claim_id: str,
+    db: DbSession,
+    current_user: CurrentUser,
+) -> ExpenseCaseTimelineRead:
+    claim = ExpenseClaimService(db).get_claim(claim_id, current_user)
+    if claim is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="费用单据不存在。")
+
+    expense_case = ExpenseCaseService(db).get_timeline_for_claim(
+        claim.id,
+        tenant_id=current_user.tenant_id,
+    )
+    if expense_case is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="该单据尚未纳入统一费用事件。",
+        )
+    return ExpenseCaseTimelineRead.model_validate(expense_case)

server/src/app/api/v1/router.py

@@ -12,6 +12,7 @@ from app.api.v1.endpoints.auth import router as auth_router
 from app.api.v1.endpoints.bootstrap import router as bootstrap_router
 from app.api.v1.endpoints.budgets import router as budgets_router
 from app.api.v1.endpoints.employees import router as employees_router
+from app.api.v1.endpoints.expense_cases import router as expense_cases_router
 from app.api.v1.endpoints.employee_profiles import router as employee_profiles_router
 from app.api.v1.endpoints.health import router as health_router
 from app.api.v1.endpoints.knowledge import router as knowledge_router
@@ -48,6 +49,7 @@ router.include_router(ontology_router, tags=["ontology"])
 router.include_router(orchestrator_router, tags=["orchestrator"])
 router.include_router(receipt_folder_router, tags=["receipt-folder"])
 router.include_router(employees_router, prefix="/employees", tags=["employees"])
+router.include_router(expense_cases_router, tags=["expense-cases"])
 router.include_router(employee_profiles_router, tags=["employee-profiles"])
 router.include_router(reimbursements_router, prefix="/reimbursements", tags=["reimbursements"])
 router.include_router(risk_observations_router, tags=["risk-observations"])

server/src/app/db/base.py

@@ -15,6 +15,7 @@ from app.models.budget import BudgetAllocation, BudgetReservation, BudgetTransac
 from app.models.employee_change_log import EmployeeChangeLog
 from app.models.employee_behavior_profile import EmployeeBehaviorProfileSnapshot
 from app.models.employee import Employee
+from app.models.expense_case import BusinessEvent, ExpenseCase, ExpenseCaseLink
 from app.models.few_shot_sample import FewShotSample
 from app.models.financial_record import (
     AccountsPayableRecord,
@@ -56,6 +57,9 @@ __all__ = [
     "BudgetReservation",
     "BudgetTransaction",
     "Employee",
+    "ExpenseCase",
+    "ExpenseCaseLink",
+    "BusinessEvent",
     "EmployeeBehaviorProfileSnapshot",
     "EmployeeChangeLog",
     "ExpenseClaim",

server/src/app/models/__init__.py

@@ -8,6 +8,7 @@ from app.models.budget import BudgetAllocation, BudgetReservation, BudgetTransac
 from app.models.employee_change_log import EmployeeChangeLog
 from app.models.employee_behavior_profile import EmployeeBehaviorProfileSnapshot
 from app.models.employee import Employee
+from app.models.expense_case import BusinessEvent, ExpenseCase, ExpenseCaseLink
 from app.models.few_shot_sample import FewShotSample
 from app.models.financial_record import (
     AccountsPayableRecord,
@@ -47,6 +48,9 @@ __all__ = [
     "BudgetReservation",
     "BudgetTransaction",
     "Employee",
+    "ExpenseCase",
+    "ExpenseCaseLink",
+    "BusinessEvent",
     "EmployeeBehaviorProfileSnapshot",
     "EmployeeChangeLog",
     "ExpenseClaim",

server/src/app/models/expense_case.py

@@ -0,0 +1,112 @@
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+from typing import Any
+
+from sqlalchemy import DateTime, ForeignKey, Index, Integer, String, Text, UniqueConstraint, func
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+from sqlalchemy.types import JSON
+
+from app.db.base_class import Base
+
+
+def _new_id() -> str:
+    return str(uuid.uuid4())
+
+
+class ExpenseCase(Base):
+    __tablename__ = "expense_cases"
+    __table_args__ = (
+        UniqueConstraint("tenant_id", "case_no", name="uq_expense_cases_tenant_case_no"),
+        Index("ix_expense_cases_tenant_stage", "tenant_id", "current_stage"),
+        Index("ix_expense_cases_tenant_status", "tenant_id", "status"),
+    )
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=_new_id)
+    tenant_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    case_no: Mapped[str] = mapped_column(String(80), nullable=False)
+    scene_code: Mapped[str] = mapped_column(String(50), nullable=False, default="other")
+    title: Mapped[str] = mapped_column(String(200), nullable=False)
+    owner_employee_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True)
+    current_stage: Mapped[str] = mapped_column(String(40), nullable=False, default="claiming")
+    status: Mapped[str] = mapped_column(String(30), nullable=False, default="active")
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
+    )
+
+    links = relationship(
+        "ExpenseCaseLink",
+        back_populates="expense_case",
+        cascade="all, delete-orphan",
+        order_by="asc(ExpenseCaseLink.created_at)",
+    )
+    events = relationship(
+        "BusinessEvent",
+        back_populates="expense_case",
+        cascade="all, delete-orphan",
+        order_by="asc(BusinessEvent.occurred_at)",
+    )
+
+
+class ExpenseCaseLink(Base):
+    __tablename__ = "expense_case_links"
+    __table_args__ = (
+        UniqueConstraint("resource_type", "resource_id", name="uq_expense_case_links_resource"),
+        Index("ix_expense_case_links_tenant_case", "tenant_id", "expense_case_id"),
+    )
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=_new_id)
+    tenant_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    expense_case_id: Mapped[str] = mapped_column(
+        ForeignKey("expense_cases.id", ondelete="CASCADE"), nullable=False
+    )
+    resource_type: Mapped[str] = mapped_column(String(50), nullable=False)
+    resource_id: Mapped[str] = mapped_column(String(100), nullable=False)
+    relation_type: Mapped[str] = mapped_column(String(50), nullable=False)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
+
+    expense_case = relationship("ExpenseCase", back_populates="links")
+
+
+class BusinessEvent(Base):
+    __tablename__ = "business_events"
+    __table_args__ = (
+        UniqueConstraint(
+            "tenant_id",
+            "aggregate_type",
+            "aggregate_id",
+            "event_type",
+            "idempotency_key",
+            name="uq_business_event_idempotency",
+        ),
+        Index("ix_business_events_tenant_case_time", "tenant_id", "expense_case_id", "occurred_at"),
+        Index("ix_business_events_outbox", "delivery_status", "occurred_at"),
+        Index("ix_business_events_aggregate", "aggregate_type", "aggregate_id"),
+    )
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=_new_id)
+    tenant_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    expense_case_id: Mapped[str] = mapped_column(
+        ForeignKey("expense_cases.id", ondelete="CASCADE"), nullable=False
+    )
+    aggregate_type: Mapped[str] = mapped_column(String(50), nullable=False)
+    aggregate_id: Mapped[str] = mapped_column(String(100), nullable=False)
+    event_type: Mapped[str] = mapped_column(String(80), nullable=False, index=True)
+    event_version: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
+    idempotency_key: Mapped[str] = mapped_column(String(120), nullable=False)
+    correlation_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    causation_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    actor_id: Mapped[str] = mapped_column(String(120), nullable=False)
+    actor_type: Mapped[str] = mapped_column(String(30), nullable=False, default="user")
+    payload_json: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False, default=dict)
+    delivery_status: Mapped[str] = mapped_column(String(20), nullable=False, default="pending")
+    delivery_attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    last_delivery_error: Mapped[str | None] = mapped_column(Text(), nullable=True)
+    occurred_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now()
+    )
+    published_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+
+    expense_case = relationship("ExpenseCase", back_populates="events")

server/src/app/repositories/audit_log.py

@@ -50,8 +50,11 @@ class AuditLogRepository:
             stmt = stmt.limit(limit)
         return list(self.db.scalars(stmt).all())
 
-    def create(self, log: AuditLog) -> AuditLog:
+    def create(self, log: AuditLog, *, commit: bool = True) -> AuditLog:
         self.db.add(log)
-        self.db.commit()
+        if commit:
+            self.db.commit()
+        else:
+            self.db.flush()
         self.db.refresh(log)
         return log

server/src/app/schemas/expense_case.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class ExpenseCaseLinkRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: str
+    resource_type: str
+    resource_id: str
+    relation_type: str
+    created_at: datetime
+
+
+class BusinessEventRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: str
+    event_type: str
+    event_version: int
+    idempotency_key: str
+    aggregate_type: str
+    aggregate_id: str
+    correlation_id: str
+    causation_id: str | None
+    actor_id: str
+    actor_type: str
+    payload_json: dict[str, Any] = Field(default_factory=dict)
+    delivery_status: str
+    occurred_at: datetime
+    published_at: datetime | None
+
+
+class ExpenseCaseTimelineRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: str
+    case_no: str
+    scene_code: str
+    title: str
+    owner_employee_id: str | None
+    current_stage: str
+    status: str
+    created_at: datetime
+    updated_at: datetime
+    links: list[ExpenseCaseLinkRead] = Field(default_factory=list)
+    events: list[BusinessEventRead] = Field(default_factory=list)

server/src/app/services/agent_foundation.py

@@ -38,6 +38,11 @@ from app.services.agent_foundation_spreadsheets import AgentFoundationSpreadshee
 logger = get_logger("app.services.agent_foundation")
 _foundation_ready_lock = threading.RLock()
 _foundation_ready_keys: set[str] = set()
+MIGRATION_OWNED_TABLES = {
+    "expense_cases",
+    "expense_case_links",
+    "business_events",
+}
 
 
 def prepare_agent_foundation() -> None:
@@ -77,7 +82,12 @@ class AgentFoundationService(
 
     def _prepare_foundation(self) -> None:
         try:
-            Base.metadata.create_all(bind=self.db.get_bind())
+            legacy_bootstrap_tables = [
+                table
+                for table in Base.metadata.sorted_tables
+                if table.name not in MIGRATION_OWNED_TABLES
+            ]
+            Base.metadata.create_all(bind=self.db.get_bind(), tables=legacy_bootstrap_tables)
             self._ensure_agent_asset_schema()
             self._ensure_financial_record_schema()
             self._seed_agent_assets()

server/src/app/services/audit.py

@@ -47,6 +47,7 @@ class AuditLogService:
         before_json: dict[str, Any] | None = None,
         after_json: dict[str, Any] | None = None,
         request_id: str | None = None,
+        commit: bool = True,
     ) -> AuditLog:
         log = AuditLog(
             actor=actor,
@@ -58,7 +59,7 @@ class AuditLogService:
             request_id=request_id or uuid.uuid4().hex,
             created_at=datetime.now(UTC),
         )
-        created = self.repository.create(log)
+        created = self.repository.create(log, commit=commit)
         logger.info(
             "Created audit log id=%s action=%s resource=%s:%s",
             created.id,

server/src/app/services/expense_cases.py

@@ -0,0 +1,306 @@
+from __future__ import annotations
+
+import hashlib
+import uuid
+from datetime import UTC, datetime
+from decimal import Decimal
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session, selectinload
+
+from app.models.expense_case import BusinessEvent, ExpenseCase, ExpenseCaseLink
+from app.models.financial_record import ExpenseClaim
+from app.services.document_numbering import is_application_claim_no
+
+DEFAULT_TENANT_ID = "default"
+
+
+class ExpenseCaseService:
+    """费用事件编排的最小持久化边界。
+
+    该服务只允许 add/flush，不负责 commit。调用方必须让业务状态与事件在同一事务中提交。
+    """
+
+    def __init__(self, db: Session) -> None:
+        self.db = db
+
+    @staticmethod
+    def normalize_tenant_id(value: str | None) -> str:
+        return str(value or DEFAULT_TENANT_ID).strip() or DEFAULT_TENANT_ID
+
+    @staticmethod
+    def normalize_correlation_id(value: str | None) -> str:
+        text = str(value or "").strip()
+        if not text:
+            return uuid.uuid4().hex
+        if len(text) <= 64:
+            return text
+        digest = hashlib.sha256(text.encode("utf-8")).hexdigest()
+        return f"sha256:{digest[:57]}"
+
+    def ensure_case_for_claim(
+        self,
+        claim: ExpenseClaim,
+        *,
+        tenant_id: str | None = None,
+        relation_type: str | None = None,
+    ) -> ExpenseCase:
+        if not claim.id:
+            self.db.flush()
+        if not claim.id:
+            raise ValueError("费用单据尚未生成 ID，无法关联费用事件。")
+
+        normalized_tenant = self.normalize_tenant_id(tenant_id)
+        existing_link = self.db.scalar(
+            select(ExpenseCaseLink).where(
+                ExpenseCaseLink.resource_type == "expense_claim",
+                ExpenseCaseLink.resource_id == claim.id,
+            )
+        )
+        if existing_link is not None:
+            if existing_link.tenant_id != normalized_tenant:
+                raise PermissionError("费用单据已属于其他租户。")
+            expense_case = self.db.get(ExpenseCase, existing_link.expense_case_id)
+            if expense_case is None or expense_case.tenant_id != normalized_tenant:
+                raise RuntimeError("费用事件关联已损坏。")
+            return expense_case
+
+        case_no = f"CASE-{str(claim.claim_no or claim.id).strip()}"
+        expense_case = self.db.scalar(
+            select(ExpenseCase).where(
+                ExpenseCase.tenant_id == normalized_tenant,
+                ExpenseCase.case_no == case_no,
+            )
+        )
+        if expense_case is None:
+            expense_case = ExpenseCase(
+                id=str(uuid.uuid4()),
+                tenant_id=normalized_tenant,
+                case_no=case_no,
+                scene_code=self._scene_code(claim),
+                title=self._case_title(claim),
+                owner_employee_id=claim.employee_id,
+                current_stage=self._stage_for_claim(claim),
+                status=self._case_status_for_claim(claim),
+            )
+            self.db.add(expense_case)
+            self.db.flush()
+
+        self.link_claim(
+            expense_case,
+            claim,
+            tenant_id=normalized_tenant,
+            relation_type=relation_type or self._relation_type(claim),
+        )
+        return expense_case
+
+    def link_claim(
+        self,
+        expense_case: ExpenseCase,
+        claim: ExpenseClaim,
+        *,
+        tenant_id: str | None = None,
+        relation_type: str,
+    ) -> ExpenseCaseLink:
+        if not claim.id:
+            self.db.flush()
+        normalized_tenant = self.normalize_tenant_id(tenant_id or expense_case.tenant_id)
+        if expense_case.tenant_id != normalized_tenant:
+            raise PermissionError("不能把费用单据关联到其他租户的费用事件。")
+
+        existing_link = self.db.scalar(
+            select(ExpenseCaseLink).where(
+                ExpenseCaseLink.resource_type == "expense_claim",
+                ExpenseCaseLink.resource_id == claim.id,
+            )
+        )
+        if existing_link is not None:
+            if (
+                existing_link.tenant_id != normalized_tenant
+                or existing_link.expense_case_id != expense_case.id
+            ):
+                raise PermissionError("费用单据已经关联到其他费用事件。")
+            return existing_link
+
+        link = ExpenseCaseLink(
+            id=str(uuid.uuid4()),
+            tenant_id=normalized_tenant,
+            expense_case_id=expense_case.id,
+            resource_type="expense_claim",
+            resource_id=claim.id,
+            relation_type=relation_type,
+        )
+        self.db.add(link)
+        self.db.flush()
+        return link
+
+    def record_claim_event(
+        self,
+        claim: ExpenseClaim,
+        *,
+        event_type: str,
+        actor_id: str,
+        tenant_id: str | None = None,
+        correlation_id: str | None = None,
+        idempotency_key: str | None = None,
+        causation_id: str | None = None,
+        previous_status: str | None = None,
+        previous_approval_stage: str | None = None,
+        extra_payload: dict[str, Any] | None = None,
+        expense_case: ExpenseCase | None = None,
+        relation_type: str | None = None,
+        update_case_state: bool = True,
+    ) -> tuple[ExpenseCase, BusinessEvent]:
+        normalized_tenant = self.normalize_tenant_id(tenant_id)
+        if expense_case is None:
+            expense_case = self.ensure_case_for_claim(
+                claim,
+                tenant_id=normalized_tenant,
+                relation_type=relation_type,
+            )
+        else:
+            self.link_claim(
+                expense_case,
+                claim,
+                tenant_id=normalized_tenant,
+                relation_type=relation_type or self._relation_type(claim),
+            )
+
+        if update_case_state:
+            expense_case.current_stage = self._stage_for_claim(claim)
+            expense_case.status = self._case_status_for_claim(claim)
+        if claim.employee_id and not expense_case.owner_employee_id:
+            expense_case.owner_employee_id = claim.employee_id
+
+        payload: dict[str, Any] = {
+            "claim_no": str(claim.claim_no or ""),
+            "expense_type": str(claim.expense_type or ""),
+            "amount": self._money_text(claim.amount),
+            "currency": str(claim.currency or "CNY"),
+            "previous_status": str(previous_status or ""),
+            "previous_approval_stage": str(previous_approval_stage or ""),
+            "next_status": str(claim.status or ""),
+            "next_approval_stage": str(claim.approval_stage or ""),
+        }
+        payload.update(extra_payload or {})
+
+        normalized_correlation_id = self.normalize_correlation_id(correlation_id)
+        normalized_idempotency_key = self._normalize_idempotency_key(
+            idempotency_key or normalized_correlation_id
+        )
+        existing_event = self.db.scalar(
+            select(BusinessEvent).where(
+                BusinessEvent.tenant_id == normalized_tenant,
+                BusinessEvent.aggregate_type == "expense_claim",
+                BusinessEvent.aggregate_id == claim.id,
+                BusinessEvent.event_type == str(event_type).strip(),
+                BusinessEvent.idempotency_key == normalized_idempotency_key,
+            )
+        )
+        if existing_event is not None:
+            return expense_case, existing_event
+
+        event = BusinessEvent(
+            id=str(uuid.uuid4()),
+            tenant_id=normalized_tenant,
+            expense_case_id=expense_case.id,
+            aggregate_type="expense_claim",
+            aggregate_id=claim.id,
+            event_type=str(event_type).strip(),
+            event_version=1,
+            idempotency_key=normalized_idempotency_key,
+            correlation_id=normalized_correlation_id,
+            causation_id=self.normalize_correlation_id(causation_id) if causation_id else None,
+            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",
+            occurred_at=datetime.now(UTC),
+        )
+        self.db.add(event)
+        self.db.flush()
+        return expense_case, event
+
+    @staticmethod
+    def _normalize_idempotency_key(value: str) -> str:
+        text = str(value or "").strip()
+        if not text:
+            raise ValueError("业务事件必须提供幂等键。")
+        if len(text) <= 120:
+            return text
+        digest = hashlib.sha256(text.encode("utf-8")).hexdigest()
+        return f"sha256:{digest}"
+
+    def get_timeline_for_claim(
+        self,
+        claim_id: str,
+        *,
+        tenant_id: str | None = None,
+    ) -> ExpenseCase | None:
+        normalized_tenant = self.normalize_tenant_id(tenant_id)
+        link = self.db.scalar(
+            select(ExpenseCaseLink).where(
+                ExpenseCaseLink.resource_type == "expense_claim",
+                ExpenseCaseLink.resource_id == claim_id,
+            )
+        )
+        if link is None or link.tenant_id != normalized_tenant:
+            return None
+        return self.db.scalar(
+            select(ExpenseCase)
+            .options(selectinload(ExpenseCase.links), selectinload(ExpenseCase.events))
+            .where(
+                ExpenseCase.id == link.expense_case_id,
+                ExpenseCase.tenant_id == normalized_tenant,
+            )
+        )
+
+    @staticmethod
+    def _money_text(value: Decimal | None) -> str:
+        return f"{Decimal(value or Decimal('0.00')).quantize(Decimal('0.01')):.2f}"
+
+    @staticmethod
+    def _scene_code(claim: ExpenseClaim) -> str:
+        expense_type = str(claim.expense_type or "").strip().lower()
+        return expense_type.removesuffix("_application") or "other"
+
+    @staticmethod
+    def _case_title(claim: ExpenseClaim) -> str:
+        reason = str(claim.reason or "").strip()
+        return reason[:200] or f"费用事件 {claim.claim_no}"
+
+    @classmethod
+    def _relation_type(cls, claim: ExpenseClaim) -> str:
+        claim_no = str(claim.claim_no or "").strip()
+        expense_type = str(claim.expense_type or "").strip().lower()
+        is_application = (
+            is_application_claim_no(claim_no)
+            or expense_type == "application"
+            or expense_type.endswith("_application")
+        )
+        return "application" if is_application else "claim"
+
+    @classmethod
+    def _stage_for_claim(cls, claim: ExpenseClaim) -> str:
+        status = str(claim.status or "").strip().lower()
+        if status in {"pending_payment"}:
+            return "paying"
+        if status in {"paid"}:
+            return "accounting"
+        if status in {"completed"}:
+            return "closed"
+        if status in {"submitted", "approved"}:
+            if cls._relation_type(claim) == "application" and status == "approved":
+                return "approved_to_spend"
+            return "reviewing"
+        return "claiming"
+
+    @staticmethod
+    def _case_status_for_claim(claim: ExpenseClaim) -> str:
+        status = str(claim.status or "").strip().lower()
+        if status in {"cancelled", "voided", "rejected"}:
+            return "cancelled"
+        if status == "completed":
+            return "closed"
+        return "active"

server/src/app/services/expense_claim_application_handoff.py

@@ -238,6 +238,10 @@ class ExpenseClaimApplicationHandoffMixin:
                 {
                     "application_claim_id": application_claim.id,
                     "application_claim_no": str(application_claim.claim_no or "").strip(),
+                    "archive_event_id": str(archive_flag.get("archive_event_id") or ""),
+                    "previous_status": previous_status,
+                    "previous_approval_stage": previous_stage,
+                    "next_status": "approved",
                     "next_approval_stage": APPLICATION_ARCHIVE_STAGE,
                 }
             )
@@ -248,6 +252,7 @@ class ExpenseClaimApplicationHandoffMixin:
                 resource_id=application_claim.id,
                 before_json=before_json,
                 after_json=self._serialize_claim(application_claim),
+                commit=False,
             )
 
         return archived_applications
@@ -396,6 +401,7 @@ class ExpenseClaimApplicationHandoffMixin:
                 resource_id=application_claim.id,
                 before_json=before_json,
                 after_json=self._serialize_claim(application_claim),
+                commit=False,
             )
 
         return synced_applications

server/src/app/services/expense_claim_approval_flow.py

@@ -6,6 +6,7 @@ from decimal import Decimal, InvalidOperation
 from typing import Any
 
 from app.api.deps import CurrentUserContext
+from app.models.financial_record import ExpenseClaim
 from app.services.budget import BudgetService
 from app.services.expense_claim_workflow_constants import (
     APPLICATION_LINK_STATUS_STAGE,
@@ -45,6 +46,7 @@ class ExpenseClaimApprovalFlowMixin:
         next_budget_manager = None
         merged_budget_approval = False
         route_decision_flag: dict[str, Any] | None = None
+        generated_draft = None
         if previous_stage == DIRECT_MANAGER_APPROVAL_STAGE:
             if not self._access_policy.can_approve_claim(current_user, claim):
                 raise ValueError("只有当前直属领导审批人可以审批通过该单据。")
@@ -254,6 +256,48 @@ class ExpenseClaimApprovalFlowMixin:
             business_stage=business_stage,
         )
 
+        correlation_id = str(approval_flag.get("approval_event_id") or uuid.uuid4())
+        structured_event_type = "approval_stage_completed"
+        if is_application_claim and next_status == "approved":
+            structured_event_type = "application_approved"
+        elif not is_application_claim and next_status == PAYMENT_PENDING_STATUS:
+            structured_event_type = "claim_approved"
+        expense_case, _event = self._expense_cases.record_claim_event(
+            claim,
+            event_type=structured_event_type,
+            actor_id=current_user.username,
+            tenant_id=getattr(current_user, "tenant_id", None),
+            correlation_id=correlation_id,
+            idempotency_key=correlation_id,
+            previous_status=str(before_json.get("status") or ""),
+            previous_approval_stage=previous_stage,
+            extra_payload={
+                "workflow_event_type": event_type,
+                "opinion": approval_opinion,
+                "route_requires_budget_review": bool(
+                    route_decision_flag and route_decision_flag.get("requires_budget_review")
+                ),
+            },
+        )
+        if generated_draft is not None:
+            self._expense_cases.record_claim_event(
+                generated_draft,
+                event_type="reimbursement_draft_generated",
+                actor_id="system",
+                tenant_id=getattr(current_user, "tenant_id", None),
+                correlation_id=correlation_id,
+                idempotency_key=correlation_id,
+                causation_id=_event.id,
+                previous_status="",
+                previous_approval_stage="",
+                extra_payload={
+                    "application_claim_id": claim.id,
+                    "application_claim_no": claim.claim_no,
+                },
+                expense_case=expense_case,
+                relation_type="generated_reimbursement",
+            )
+
         self.db.commit()
         self.db.refresh(claim)
         self._access_policy.attach_budget_approval_snapshot(claim)
@@ -326,6 +370,47 @@ class ExpenseClaimApprovalFlowMixin:
         claim.approval_stage = PAYMENT_PAID_STAGE
         claim.risk_flags_json = [*list(claim.risk_flags_json or []), payment_flag]
 
+        payment_correlation_id = str(payment_flag.get("payment_event_id") or uuid.uuid4())
+        expense_case, payment_event = self._expense_cases.record_claim_event(
+            claim,
+            event_type="payment_completed",
+            actor_id=current_user.username,
+            tenant_id=getattr(current_user, "tenant_id", None),
+            correlation_id=payment_correlation_id,
+            idempotency_key=payment_correlation_id,
+            previous_status=str(before_json.get("status") or ""),
+            previous_approval_stage=previous_stage,
+            extra_payload={"archived_applications": archived_applications},
+        )
+        for archived_application in archived_applications:
+            application_claim = self.db.get(
+                ExpenseClaim,
+                str(archived_application.get("application_claim_id") or ""),
+            )
+            if application_claim is None:
+                raise RuntimeError("付款关联的申请单已不存在，无法记录归档事件。")
+            archive_event_id = str(archived_application.get("archive_event_id") or "").strip()
+            self._expense_cases.record_claim_event(
+                application_claim,
+                event_type="application_archived",
+                actor_id=current_user.username,
+                tenant_id=getattr(current_user, "tenant_id", None),
+                correlation_id=payment_correlation_id,
+                idempotency_key=archive_event_id,
+                causation_id=payment_event.id,
+                previous_status=str(archived_application.get("previous_status") or ""),
+                previous_approval_stage=str(
+                    archived_application.get("previous_approval_stage") or ""
+                ),
+                extra_payload={
+                    "reimbursement_claim_id": claim.id,
+                    "reimbursement_claim_no": claim.claim_no,
+                },
+                expense_case=expense_case,
+                relation_type="application",
+                update_case_state=False,
+            )
+
         self.db.commit()
         self.db.refresh(claim)
 

server/src/app/services/expense_claim_draft_flow.py

@@ -1020,6 +1020,15 @@ class ExpenseClaimDraftFlowMixin(ExpenseClaimApplicationLinkMixin, ExpenseClaimD
                 self._sync_claim_from_items(claim)
             if locked_expense_type:
                 claim.expense_type = locked_expense_type
+            self._expense_cases.record_claim_event(
+                claim,
+                event_type=("claim_draft_created" if is_new_claim else "claim_draft_updated"),
+                actor_id=user_id or claim.employee_name or "system",
+                correlation_id=run_id,
+                idempotency_key=run_id,
+                previous_status=str((before_json or {}).get("status") or ""),
+                previous_approval_stage=str((before_json or {}).get("approval_stage") or ""),
+            )
             self.db.commit()
             self.db.refresh(claim)
         except IntegrityError as exc:
@@ -1066,4 +1075,3 @@ class ExpenseClaimDraftFlowMixin(ExpenseClaimApplicationLinkMixin, ExpenseClaimD
             "amount": float(claim.amount),
             "invoice_count": int(claim.invoice_count or 0),
         }
-

server/src/app/services/expense_claims.py

@@ -115,6 +115,7 @@ from app.services.expense_claim_constants import (
     TRAVEL_POLICY_HOTEL_NIGHT_PATTERN,
     STANDARD_ADJUSTMENT_RISK_SOURCE,
 )
+from app.services.expense_cases import ExpenseCaseService
 from app.services.expense_claim_risk_review import ExpenseClaimRiskReviewMixin
 from app.services.expense_amounts import (
     extract_amount_candidates,
@@ -645,6 +646,20 @@ class ExpenseClaimItemActionMixin:
 
         claim.risk_flags_json = dedupe_claim_risk_flags(claim.risk_flags_json)
 
+        self._expense_cases.record_claim_event(
+            claim,
+            event_type=("application_submitted" if is_application_claim else "claim_submitted"),
+            actor_id=current_user.username,
+            tenant_id=getattr(current_user, "tenant_id", None),
+            idempotency_key=(
+                f"submit:{claim.id}:{claim.submitted_at.isoformat()}"
+                if claim.submitted_at is not None
+                else f"submit:{claim.id}:{before_json.get('status') or 'draft'}"
+            ),
+            previous_status=str(before_json.get("status") or ""),
+            previous_approval_stage=str(before_json.get("approval_stage") or ""),
+        )
+
         self.db.commit()
         self.db.refresh(claim)
 
@@ -843,6 +858,20 @@ class ExpenseClaimItemActionMixin:
             business_stage="expense_application" if is_application_claim else "reimbursement",
         )
 
+        self._expense_cases.record_claim_event(
+            claim,
+            event_type=("application_returned" if is_application_claim else "claim_returned"),
+            actor_id=current_user.username,
+            tenant_id=getattr(current_user, "tenant_id", None),
+            idempotency_key=str(return_flag.get("return_event_id") or ""),
+            previous_status=previous_status,
+            previous_approval_stage=previous_stage,
+            extra_payload={
+                "reason": message,
+                "reason_codes": normalized_reason_codes,
+            },
+        )
+
         self.db.commit()
         self.db.refresh(claim)
 
@@ -862,6 +891,7 @@ class ExpenseClaimService(ExpenseClaimStandardAdjustmentMixin, ExpenseClaimItemA
     def __init__(self, db: Session) -> None:
         self.db = db
         self.audit_service = AuditLogService(db)
+        self._expense_cases = ExpenseCaseService(db)
         self._access_policy = ExpenseClaimAccessPolicy(db)
         self._attachment_storage = ExpenseClaimAttachmentStorage()
         self._attachment_presentation = ExpenseClaimAttachmentPresentation(self._attachment_storage)

server/tests/test_expense_case_service.py

@@ -0,0 +1,423 @@
+from __future__ import annotations
+
+from datetime import UTC, date, datetime
+from decimal import Decimal
+
+import pytest
+from sqlalchemy import create_engine, inspect, select
+from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.pool import StaticPool
+
+from app.api.deps import CurrentUserContext
+from app.db.base import Base
+from app.models.audit_log import AuditLog
+from app.models.budget import BudgetAllocation
+from app.models.employee import Employee
+from app.models.expense_case import BusinessEvent, ExpenseCase, ExpenseCaseLink
+from app.models.financial_record import ExpenseClaim, ExpenseClaimItem
+from app.models.organization import OrganizationUnit
+from app.services.agent_foundation import AgentFoundationService
+from app.services.expense_cases import ExpenseCaseService
+from app.services.expense_claim_workflow_constants import (
+    APPLICATION_ARCHIVE_STAGE,
+    APPLICATION_LINK_STATUS_STAGE,
+    DIRECT_MANAGER_APPROVAL_STAGE,
+)
+from app.services.expense_claims import ExpenseClaimService
+
+
+def build_session() -> Session:
+    engine = create_engine(
+        "sqlite+pysqlite:///:memory:",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
+    Base.metadata.create_all(bind=engine)
+    session_factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)
+    return session_factory()
+
+
+def build_claim(
+    *,
+    claim_no: str,
+    employee: Employee | None = None,
+    status: str = "draft",
+    approval_stage: str = "待提交",
+    expense_type: str = "transport",
+) -> ExpenseClaim:
+    claim = ExpenseClaim(
+        claim_no=claim_no,
+        employee_id=employee.id if employee is not None else None,
+        employee_name=employee.name if employee is not None else "张三",
+        department_name="市场部",
+        project_code="PRJ-CASE",
+        expense_type=expense_type,
+        reason="客户现场差旅",
+        location="上海",
+        amount=Decimal("88.00"),
+        currency="CNY",
+        invoice_count=1,
+        occurred_at=datetime(2026, 7, 13, 9, 0, tzinfo=UTC),
+        submitted_at=(datetime(2026, 7, 13, 10, 0, tzinfo=UTC) if status != "draft" else None),
+        status=status,
+        approval_stage=approval_stage,
+        risk_flags_json=[],
+    )
+    if status == "draft":
+        claim.items = [
+            ExpenseClaimItem(
+                item_date=date(2026, 7, 13),
+                item_type="transport",
+                item_reason="客户现场交通",
+                item_location="上海",
+                item_note="",
+                item_amount=Decimal("88.00"),
+                invoice_id="invoice-case-1",
+            )
+        ]
+    return claim
+
+
+def test_event_write_uses_caller_transaction_and_tenant_scope() -> None:
+    with build_session() as db:
+        claim = build_claim(claim_no="RE-CASE-ROLLBACK")
+        db.add(claim)
+        db.commit()
+
+        service = ExpenseCaseService(db)
+        service.record_claim_event(
+            claim,
+            event_type="claim_draft_created",
+            actor_id="owner@example.com",
+            tenant_id="tenant-a",
+            correlation_id="run-case-rollback",
+        )
+
+        assert db.scalar(select(ExpenseCase)) is not None
+        assert db.scalar(select(BusinessEvent)) is not None
+        db.rollback()
+
+        assert db.scalar(select(ExpenseCase)) is None
+        assert db.scalar(select(ExpenseCaseLink)) is None
+        assert db.scalar(select(BusinessEvent)) is None
+
+        service.record_claim_event(
+            claim,
+            event_type="claim_draft_created",
+            actor_id="owner@example.com",
+            tenant_id="tenant-a",
+        )
+        db.commit()
+
+        assert service.get_timeline_for_claim(claim.id, tenant_id="tenant-a") is not None
+        assert service.get_timeline_for_claim(claim.id, tenant_id="tenant-b") is None
+        with pytest.raises(PermissionError, match="其他租户"):
+            service.record_claim_event(
+                claim,
+                event_type="claim_draft_updated",
+                actor_id="other@example.com",
+                tenant_id="tenant-b",
+            )
+
+
+def test_legacy_bootstrap_excludes_migration_owned_tables(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    engine = create_engine("sqlite+pysqlite:///:memory:")
+    session_factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)
+    with session_factory() as db:
+        service = AgentFoundationService(db)
+        monkeypatch.setattr(service, "_ensure_agent_asset_schema", lambda: None)
+        monkeypatch.setattr(service, "_ensure_financial_record_schema", lambda: None)
+        monkeypatch.setattr(service, "_seed_agent_assets", lambda: None)
+        monkeypatch.setattr(service, "_sync_demo_financial_records", lambda: None)
+        monkeypatch.setattr(service, "_seed_runs_and_logs", lambda: None)
+
+        service._prepare_foundation()
+
+    table_names = set(inspect(engine).get_table_names())
+    assert "employees" in table_names
+    assert {"expense_cases", "expense_case_links", "business_events"}.isdisjoint(table_names)
+
+
+def test_event_write_is_idempotent_for_same_business_operation() -> None:
+    with build_session() as db:
+        claim = build_claim(claim_no="RE-CASE-IDEMPOTENT")
+        db.add(claim)
+        db.commit()
+
+        service = ExpenseCaseService(db)
+        _case, first_event = service.record_claim_event(
+            claim,
+            event_type="claim_draft_updated",
+            actor_id="owner@example.com",
+            tenant_id="tenant-a",
+            correlation_id="run-idempotent",
+            idempotency_key="save-operation-1",
+        )
+        _case, repeated_event = service.record_claim_event(
+            claim,
+            event_type="claim_draft_updated",
+            actor_id="owner@example.com",
+            tenant_id="tenant-a",
+            correlation_id="run-idempotent",
+            idempotency_key="save-operation-1",
+        )
+        db.commit()
+
+        assert repeated_event.id == first_event.id
+        assert len(list(db.scalars(select(BusinessEvent)).all())) == 1
+
+
+def test_submit_claim_creates_case_link_and_structured_event() -> None:
+    current_user = CurrentUserContext(
+        username="employee-case@example.com",
+        name="张三",
+        role_codes=[],
+        is_admin=False,
+    )
+
+    with build_session() as db:
+        manager = Employee(
+            employee_no="CASE-MANAGER",
+            name="李经理",
+            email="manager-case@example.com",
+        )
+        employee = Employee(
+            employee_no="CASE-EMPLOYEE",
+            name="张三",
+            email=current_user.username,
+            manager=manager,
+        )
+        db.add_all([manager, employee])
+        db.flush()
+        claim = build_claim(claim_no="RE-CASE-SUBMIT", employee=employee)
+        claim.risk_flags_json = [
+            {
+                "source": "ai_pre_review",
+                "status": "passed",
+                "passed": True,
+                "severity": "info",
+                "blocking_risk_count": 0,
+            }
+        ]
+        db.add(claim)
+        db.commit()
+
+        submitted = ExpenseClaimService(db).submit_claim(claim.id, current_user)
+
+        assert submitted is not None
+        assert submitted.status == "submitted"
+        link = db.scalar(select(ExpenseCaseLink).where(ExpenseCaseLink.resource_id == submitted.id))
+        assert link is not None
+        event = db.scalar(select(BusinessEvent).where(BusinessEvent.aggregate_id == submitted.id))
+        assert event is not None
+        assert event.event_type == "claim_submitted"
+        assert event.delivery_status == "pending"
+        assert event.payload_json["previous_status"] == "draft"
+        assert event.payload_json["next_status"] == "submitted"
+
+
+def test_payment_event_failure_rolls_back_payment_archive_and_nested_audit(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    current_user = CurrentUserContext(
+        username="finance-case@example.com",
+        name="财务付款",
+        role_codes=["finance"],
+        is_admin=False,
+    )
+
+    with build_session() as db:
+        application_claim = build_claim(
+            claim_no="AP-CASE-ARCHIVE",
+            status="approved",
+            approval_stage="关联单据状态",
+            expense_type="travel_application",
+        )
+        reimbursement_claim = build_claim(
+            claim_no="RE-CASE-PAY",
+            status="pending_payment",
+            approval_stage="待付款",
+            expense_type="travel",
+        )
+        reimbursement_claim.risk_flags_json = [
+            {
+                "source": "application_handoff",
+                "application_claim_id": application_claim.id,
+                "application_claim_no": application_claim.claim_no,
+            }
+        ]
+        db.add_all([application_claim, reimbursement_claim])
+        db.commit()
+
+        service = ExpenseClaimService(db)
+
+        def fail_event(*args, **kwargs):
+            raise RuntimeError("outbox unavailable")
+
+        monkeypatch.setattr(service._expense_cases, "record_claim_event", fail_event)
+        with pytest.raises(RuntimeError, match="outbox unavailable"):
+            service.mark_claim_paid(reimbursement_claim.id, current_user)
+        db.rollback()
+
+        db.refresh(reimbursement_claim)
+        db.refresh(application_claim)
+        assert reimbursement_claim.status == "pending_payment"
+        assert reimbursement_claim.approval_stage == "待付款"
+        assert application_claim.status == "approved"
+        assert application_claim.approval_stage != APPLICATION_ARCHIVE_STAGE
+        assert db.scalar(select(AuditLog)) is None
+
+
+def test_application_approval_links_generated_reimbursement_to_same_case() -> None:
+    with build_session() as db:
+        department = OrganizationUnit(
+            unit_code="CASE-TRAVEL",
+            name="差旅试点部",
+            unit_type="department",
+        )
+        manager = Employee(
+            employee_no="CASE-APP-MANAGER",
+            name="差旅经理",
+            email="travel-manager@example.com",
+            organization_unit=department,
+        )
+        employee = Employee(
+            employee_no="CASE-APP-EMPLOYEE",
+            name="差旅员工",
+            email="travel-employee@example.com",
+            manager=manager,
+            organization_unit=department,
+        )
+        db.add_all([department, manager, employee])
+        db.flush()
+        db.add(
+            BudgetAllocation(
+                budget_no="BUD-CASE-TRAVEL",
+                fiscal_year=2026,
+                period_type="year",
+                period_key="2026",
+                department_id=department.id,
+                department_name=department.name,
+                cost_center=None,
+                project_code=None,
+                subject_code="travel",
+                subject_name="差旅费",
+                original_amount=Decimal("50000.00"),
+                adjusted_amount=Decimal("0.00"),
+                status="active",
+                warning_threshold=Decimal("80.00"),
+                control_action="block",
+            )
+        )
+        application_claim = build_claim(
+            claim_no="AP-CASE-GENERATE",
+            employee=employee,
+            status="submitted",
+            approval_stage=DIRECT_MANAGER_APPROVAL_STAGE,
+            expense_type="travel_application",
+        )
+        application_claim.amount = Decimal("500.00")
+        db.add(application_claim)
+        db.commit()
+
+        approved = ExpenseClaimService(db).approve_claim(
+            application_claim.id,
+            CurrentUserContext(
+                username=manager.email,
+                name=manager.name,
+                role_codes=["manager"],
+                is_admin=False,
+            ),
+            opinion="业务必要，同意申请",
+        )
+
+        assert approved is not None
+        assert approved.status == "approved"
+        assert approved.approval_stage == APPLICATION_LINK_STATUS_STAGE
+        expense_case = db.scalar(select(ExpenseCase))
+        assert expense_case is not None
+        links = list(
+            db.scalars(
+                select(ExpenseCaseLink)
+                .where(ExpenseCaseLink.expense_case_id == expense_case.id)
+                .order_by(ExpenseCaseLink.created_at)
+            ).all()
+        )
+        assert {link.relation_type for link in links} == {
+            "application",
+            "generated_reimbursement",
+        }
+        events = list(
+            db.scalars(
+                select(BusinessEvent)
+                .where(BusinessEvent.expense_case_id == expense_case.id)
+                .order_by(BusinessEvent.occurred_at)
+            ).all()
+        )
+        assert [event.event_type for event in events] == [
+            "application_approved",
+            "reimbursement_draft_generated",
+        ]
+        assert len({event.correlation_id for event in events}) == 1
+        assert events[1].causation_id == events[0].id
+
+
+def test_payment_records_application_archive_event_in_same_case() -> None:
+    current_user = CurrentUserContext(
+        username="finance-archive@example.com",
+        name="财务付款",
+        role_codes=["finance"],
+        is_admin=False,
+    )
+
+    with build_session() as db:
+        application_claim = build_claim(
+            claim_no="AP-CASE-ARCHIVE-EVENT",
+            status="approved",
+            approval_stage="关联单据状态",
+            expense_type="travel_application",
+        )
+        reimbursement_claim = build_claim(
+            claim_no="RE-CASE-PAY-EVENT",
+            status="pending_payment",
+            approval_stage="待付款",
+            expense_type="travel",
+        )
+        reimbursement_claim.risk_flags_json = [
+            {
+                "source": "application_handoff",
+                "application_claim_no": application_claim.claim_no,
+            }
+        ]
+        db.add_all([application_claim, reimbursement_claim])
+        db.commit()
+
+        paid = ExpenseClaimService(db).mark_claim_paid(reimbursement_claim.id, current_user)
+
+        assert paid is not None
+        expense_case = db.scalar(select(ExpenseCase))
+        assert expense_case is not None
+        links = list(
+            db.scalars(
+                select(ExpenseCaseLink).where(ExpenseCaseLink.expense_case_id == expense_case.id)
+            ).all()
+        )
+        assert {link.resource_id for link in links} == {
+            application_claim.id,
+            reimbursement_claim.id,
+        }
+        events = list(
+            db.scalars(
+                select(BusinessEvent)
+                .where(BusinessEvent.expense_case_id == expense_case.id)
+                .order_by(BusinessEvent.occurred_at)
+            ).all()
+        )
+        assert [event.event_type for event in events] == [
+            "payment_completed",
+            "application_archived",
+        ]
+        assert events[1].causation_id == events[0].id
+        assert expense_case.current_stage == "accounting"