feat(expenses): secure timeline and draft events

document/development/2026-07-13/dev-logs/bugs/application-draft-replay-duplicates.md

@@ -0,0 +1,8 @@
+## 修复记录
+
+- 22:59：记录 bug 修复：AI 工作台重复保存请求会创建多张申请草稿和多条事件。
+  - Git 提交检查：已执行 `git fetch --all --prune`，未发现 upstream 新提交；本地 ahead 4 个既有提交，分别为 `22669a90 feat(expenses): show unified expense event timeline`（真实费用事件时间线）、`a616b30c fix(expenses): unify AI application submission transaction`（AI 申请提交事务）、`653eda05 feat(auth): add opaque bearer sessions`（不透明 Bearer 会话）和 `661990b2 feat(expenses): add transactional expense case events`（Expense Case 事务事件基础）。
+  - 修改：新增 `expense_application_draft_events.py`，以租户、操作人、run ID 和稳定草稿快照生成固定长度幂等键及 UUID5 聚合 ID；`user_agent_application.py` 在新建前复用已完成动作，并在并发主键竞争后回滚、重查已提交事件。持久化兜底的当前时间不参与指纹，原始申请时间仍通过申请详情快照区分。
+  - 操作：先在容器中分别复现“带日期相同请求”和“缺日期半成品相同请求”均生成两个不同 `claim_id`，再补动作级幂等和数据库并发仲裁路径；业务单号继续使用既有随机格式，没有修改数据库结构或执行迁移。
+  - 验证：相同 HTTP 保存请求连续执行两次返回同一 `claim_id` 和 `claim_no`，数据库只有一张申请草稿与一条 `claim_draft_created`；同一 run 的不同快照仍分别留痕，事件失败时草稿、Case、Link 和事件整体回滚。容器内受影响后端定向回归 36 项、前端时间线兼容测试 9 项及 Python `ruff --select F,I` 通过。
+  - 影响：网络重试、用户重复点击或客户端重放不会制造重复申请草稿；并发请求由稳定聚合主键仲裁。真实 PostgreSQL 双会话并发集成测试仍是后续可补的非阻断验证，不影响当前顺序重放与事务契约。

document/development/2026-07-13/dev-logs/bugs/application-preview-forged-identity.md

@@ -0,0 +1,8 @@
+## 修复记录
+
+- 22:59：记录 bug 修复：AI 申请预览入口可保留请求体伪造的身份和管理员权限。
+  - Git 提交检查：已执行 `git fetch --all --prune`，未发现 upstream 新提交；本地 ahead 4 个既有提交，分别为 `22669a90 feat(expenses): show unified expense event timeline`（真实费用事件时间线）、`a616b30c fix(expenses): unify AI application submission transaction`（AI 申请提交事务）、`653eda05 feat(auth): add opaque bearer sessions`（不透明 Bearer 会话）和 `661990b2 feat(expenses): add transactional expense case events`（Expense Case 事务事件基础）。
+  - 修改：`reimbursements.py` 将 `user_id`、租户、角色、管理员标记、用户名、姓名、部门、职位、职级、员工编号和直属经理全部强制绑定到服务端当前会话，不再对请求体同名字段使用 `setdefault`；`test_reimbursement_endpoints.py` 增加伪造管理员与他人身份编辑退回申请的对抗用例。
+  - 操作：先在容器中复现修复前接口返回 200 且允许修改他人申请，再完成服务端身份覆盖；没有修改数据库结构，没有执行迁移或重启服务。
+  - 验证：修复后恶意请求返回 400，目标申请的事由、状态和审批节点保持不变，也没有新增费用事件；正常保存草稿与直接提交回归通过。本轮受影响后端定向回归共 36 项、前端时间线兼容测试 9 项和 Python `ruff --select F,I` 均在容器内通过。
+  - 影响：AI 工作台快速申请入口不能再通过伪造 `user_id`、`is_admin`、角色或员工编号绕过申请所有权检查，授权事实与其他受保护接口统一来自服务端会话。

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

@@ -262,6 +262,7 @@
 - Web 端只在 `sessionStorage` 保存 token 和过期时间，普通请求、流式请求及页面关闭收尾统一携带 Bearer；任一 `401` 触发本地会话清理，登出时会话指标与 token 撤销在同一事务完成。
 - 已初始化系统的 Bootstrap 状态只返回脱敏信息，Bootstrap 写入、系统设置、模型连通性、缓存、审计与系统日志等敏感管理面由平台管理员权限保护；Vite 本地 Setup 桥在初始化完成后锁定重新配置入口。
 - P0 数据契约即引入最小 `tenant_id`、数据库约束、行级过滤、向量库命名空间和对象存储前缀隔离；删除传播、数据导出和私有部署加固可在商业化阶段继续完善。
+- Expense Case 用户态查询只返回流程摘要：Case 基础状态、关系类型、事件类型、操作人、发生时间及白名单业务载荷；不返回关联资源 ID、幂等键、correlation、causation、聚合标识或 Outbox 投递状态。有权查看入口单据的本人、当前审批人、财务和管理员可查看整 Case 的安全摘要，无权限与跨租户查询继续以 404 隐藏资源存在性。
 - 自动化权限按动作、金额、场景、风险和有效期授予，不使用全局“允许 Agent 自动执行”开关。
 - 收款账户变更、资金支付、制度发布、高风险驳回和敏感主数据变更执行双人或更高等级复核。
 
@@ -443,7 +444,7 @@ docker exec -w /app -e SERVER_VENV_DIR=/tmp/x-financial-server-venv \
 - 领域模型迁移：旧 `ReimbursementRequest`、`ExpenseClaim` 和 JSON 状态并存，必须旁路记录、小步迁移和双读校验。
 - 认证和租户：当前客户端身份头不适合自动化和 SaaS，多租户、记忆和高风险动作开发前必须修复。
 - 会话运维：不透明会话已经替代客户端身份头，但仍需补充定时清理、活跃会话查看/全部退出、密钥轮换策略、登录限流和企业 SSO；当前 `tenant_id` 仍是最小契约，不代表跨租户查询守卫已经完成。
-- 费用事件读取边界：当前查询响应仍包含幂等键、correlation、causation 和 Outbox 投递字段，并在校验入口单据权限后返回整 Case 事件；进入多租户试点前必须改为用户态精简 DTO，并以 HTTP 权限测试冻结关联事件、审批意见、退回原因和操作人的可见范围。
+- 费用事件读取边界：用户态精简 DTO 和首批 HTTP 权限测试已完成；剩余风险是同一 URL 若已有外部客户端依赖旧内部字段会产生契约变更，且未来新增敏感 payload 字段必须继续显式进入白名单，不能恢复任意字典透传。
 - 反馈投毒：一次点击或违规习惯不能直接成为记忆，需要候选态、最小样本、制度约束和结果权重。
 - 自动化失控：高准确率不代表高风险动作可以无人值守，必须按动作授权并支持 shadow、Canary、抽检和回滚。
 - 虚假节省：风险暴露金额、暂缓付款和工时估算容易被夸大，必须由客户财务确认并执行去重。
@@ -494,4 +495,8 @@ docker exec -w /app -e SERVER_VENV_DIR=/tmp/x-financial-server-venv \
 - 2026-07-13（提交一致性验证）：容器内 AI 申请直接提交、失败回滚、保存草稿副作用、申请提交主链路及预算/Expense Case 回归共 21 项通过，`ruff --select F,I` 通过；未修改数据库结构，未对持久化开发数据库执行迁移。
 - 2026-07-13（P0 前端时间线）：申请/报销详情在原有横向进度下接入真实 `/api/v1/expense-cases/by-claim/{claim_id}`，新增稳定排序、中文事件语义、状态迁移、退回原因、操作人和发生时间展示；未知事件使用通用文案，不暴露 correlation、幂等键或 Outbox 投递状态。
 - 2026-07-13（前端兼容与验证）：单据切换时清空旧时间线并通过请求序列防止乱序覆盖；未纳入 Expense Case 的 404 显示非阻断兼容态，其他接口错误支持重试且不影响原有进度与单据操作。容器内 99 条定向前端测试和 Vite 生产构建通过。
+- 2026-07-13（P0 用户态事件契约）：Expense Case 响应收口为 Case 基础状态、关系类型和事件安全摘要，事件载荷采用递归白名单；关联单据 ID、聚合信息、幂等与关联链、Outbox 投递字段及未知嵌套内部字段不再通过用户接口返回。
+- 2026-07-13（P0 草稿事件补缝）：AI 工作台与小财管家新建/更新费用申请草稿分别写入 `claim_draft_created` / `claim_draft_updated`；租户从服务端身份透传，事件、Case、Link 与草稿同事务提交，失败整体回滚。动作幂等组合租户、操作人、run ID 与稳定草稿快照：完全相同的 HTTP 保存重放复用同一草稿和事件，并发竞争由稳定聚合 ID 与数据库主键仲裁，同一 run 内的真实内容变化仍保留独立事件。
+- 2026-07-13（AI 申请身份边界）：申请预览快速入口不再接受请求体中的 `user_id`、管理员标记、角色、员工编号或其他身份字段作为授权事实，全部强制绑定服务端会话；伪造管理员身份编辑他人退回申请会返回 400，且原申请与费用事件不发生变化。
+- 2026-07-13（安全与事务验证）：容器内受影响后端定向回归 36 项、Expense Case 前端兼容测试 9 项和 Python `ruff --select F,I` 通过；覆盖本人、当前审批人、财务、管理员、无权限、跨租户、整 Case 安全摘要、草稿新建/更新、失败回滚、HTTP 创建重放、相同快照去重、不同快照留痕、Steward 重放及伪造身份越权。未修改数据库结构，未执行持久化开发数据库迁移。
 - 2026-07-13（联调边界）：当前持久化开发数据库尚无 `auth_sessions`、`expense_cases`、`expense_case_links` 和 `business_events` 表，浏览器登录无法获得有效认证凭证，因此本轮未声称完成真实页面端到端联调，也未擅自执行数据库迁移。计划、消费/票据、入账、对账和复盘事件仍待后续补齐。

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

@@ -54,7 +54,8 @@
   证据：`bootstrap.py`、`settings.py`、`audit_logs.py`、`agent_traces.py`、`system_logs.py`、`vite.config.js`；已初始化 Bootstrap 脱敏且拒绝匿名重配，平台管理员/业务经理权限边界和 Vite Setup 锁测试通过。
 - [ ] [CONCEPT: 权限与安全] 继续盘点并收口风险规则发布、制度发布及其他尚未纳入本轮的敏感动作，按动作定义平台管理员或双人复核权限。
 - [ ] [CONCEPT: 权限与安全] 为所有新增表和共享核心数据补齐最小 `tenant_id`、数据库约束、查询守卫及默认租户迁移。
-- [ ] [CONCEPT: 权限与安全] 为 Expense Case 查询提供面向用户的精简事件 DTO，移除幂等键、correlation、causation 和 Outbox 投递字段，并明确审批意见、退回原因和操作人可见范围。
+- [x] [CONCEPT: 权限与安全] 为 Expense Case 查询提供面向用户的精简事件 DTO，移除幂等键、correlation、causation 和 Outbox 投递字段，并明确审批意见、退回原因和操作人可见范围。
+  证据：`expense_case.py`、`test_expense_case_endpoints.py`；用户态响应只保留安全流程摘要，关联资源 ID 和未知嵌套 payload 被递归过滤，本人、当前审批人、财务、管理员、无权限及跨租户边界测试通过。
 - [ ] [CONCEPT: 权限与安全] 为 Qdrant collection/namespace、对象存储前缀和缓存键补齐租户隔离回归测试。
 - [ ] [CONCEPT: 数据与契约] 建立 Alembic baseline 和正式迁移链，停止请求路径运行 DDL。
 - [x] [CONCEPT: 费用领域与编排] 新增 `ExpenseCaseService` 和费用事件查询接口，保持编排与具体职责分离。
@@ -67,6 +68,10 @@
   证据：`expense_claim_draft_flow.py`、`expense_claims.py`、`expense_claim_approval_flow.py`、`test_expense_case_service.py`；容器测试覆盖同 Case 关联、重复事件去重及 Outbox 失败整体回滚。
 - [x] [CONCEPT: 业务事件与 AI 决策] 收口 AI 新建申请直接提交入口，统一复用申请提交事务，并消除预算 metadata 检查对外层事务的隐式提交。
   证据：`user_agent_application.py`、`budget.py`、`test_reimbursement_endpoints.py`、`test_user_agent_service.py`；容器测试覆盖有效预算预占、`application_submitted` 事件、提交风险标记、事件失败整体回滚及保存草稿无提交副作用。
+- [x] [CONCEPT: 业务事件与 AI 决策] 为 AI 工作台和小财管家申请草稿补齐新建/更新事件、租户透传、事务回滚和动作幂等。
+  证据：`expense_application_draft_events.py`、`user_agent_application.py`、`reimbursements.py`、`steward_action_executor.py`、`test_user_agent_application_draft_events.py`、`test_reimbursement_endpoints.py`、`test_steward_action_executor.py`；完全相同 HTTP 保存重放复用同一草稿和事件，同一 run 内不同快照分别留痕，事件失败后草稿与 Case 数据整体回滚。
+- [x] [CONCEPT: 权限与安全] 将 AI 申请预览快速入口的用户、租户、角色与管理员身份强制绑定到服务端会话，拒绝请求体伪造身份编辑他人申请。
+  证据：`reimbursements.py`、`test_reimbursement_endpoints.py`；对抗用例修复前返回 200，修复后返回 400，且目标申请和费用事件保持不变。
 - [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` 映射；切换后禁止关键事件可丢弃写入。
@@ -147,7 +152,8 @@
 - [x] [CONCEPT: 测试方案] 为服务端会话、管理员保护和 Bootstrap 重配置补充首批安全回归测试。
   证据：`test_auth_session_endpoints.py`、`test_auth_service.py`、`test_bootstrap_security.py`；容器定向测试覆盖 token 摘要、伪造身份头、过期/撤销、登出原子收尾、业务经理越权和初始化后匿名重配置拒绝。
 - [ ] [CONCEPT: 测试方案] 为租户隔离、跨租户访问、规则/制度发布和双人复核等剩余敏感动作补充安全回归测试。
-- [ ] [CONCEPT: 测试方案] 为 Expense Case GET 接口补充 owner、审批人、财务、管理员、无权限用户和整 Case 关联事件可见范围的 HTTP 权限测试。
+- [x] [CONCEPT: 测试方案] 为 Expense Case GET 接口补充 owner、审批人、财务、管理员、无权限用户和整 Case 关联事件可见范围的 HTTP 权限测试。
+  证据：`test_expense_case_endpoints.py` 容器内 8 项通过，覆盖跨租户、无 Case、申请与报销关联摘要以及内部字段递归过滤。
 - [ ] [CONCEPT: 测试方案] 为 Alembic baseline、升级、旧数据迁移和回滚边界补充 Postgres 集成测试。
 - [ ] [CONCEPT: 测试方案] 为连接器幂等、重试、回执、失败恢复、重复付款和对账补充测试。
 - [ ] [CONCEPT: 测试方案] 跑通申请 → 票据 → 报销 → 预审 → 审批 → 付款 → 入账 → 归档端到端。
@@ -156,6 +162,8 @@
 - [ ] [CONCEPT: 测试方案] 跑通节省机会 → 执行 → 实现 → 财务确认 → ROI 看板闭环。
 - [x] [CONCEPT: 测试方案] 为已有 Expense Case 事件时间线补充视图模型、404 降级、详情页接入及相关响应式回归，并完成前端生产构建。
   证据：容器内 `node --test` 定向执行 99 项通过；`npm --prefix web run build` 通过。浏览器可打开本地应用，但当前未迁移数据库无法签发有效认证凭证，真实详情页联调留待迁移后完成。
+- [x] [CONCEPT: 测试方案] 为 AI 申请草稿事件补充事务失败回滚、同快照幂等、同 run 多版本留痕和 Steward 重放回归。
+  证据：本轮受影响后端定向回归 36 项、Expense Case 前端兼容测试 9 项和 Python `ruff --select F,I` 在容器内通过。
 - [ ] [CONCEPT: 测试方案] 补充其余前端组件、键盘操作、移动真实接口和完整浏览器关键流程验证。
 - [ ] [CONCEPT: 测试方案] 所有后端、集成和迁移测试在当前主应用容器内执行，单条命令最大超时 60s。
 - [ ] [CONCEPT: 指标与验收] 记录测试、lint、typecheck、构建、端到端和未覆盖风险证据。

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

@@ -104,15 +104,21 @@ def _build_application_preview_action_context(
     context_json.setdefault("entry_source", "workbench_ai_inline")
     context_json.setdefault("document_type", "expense_application")
     context_json.setdefault("application_stage", "expense_application")
-    context_json.setdefault("role_codes", current_user.role_codes)
+    # 身份与权限字段只能来自服务端会话，不能保留请求体中的同名值。
-    context_json.setdefault("is_admin", current_user.is_admin)
+    context_json.update(
-    context_json.setdefault("username", current_user.username)
+        {
-    context_json.setdefault("name", current_user.name)
+            "tenant_id": current_user.tenant_id,
-    context_json.setdefault("department_name", current_user.department_name)
+            "role_codes": current_user.role_codes,
-    context_json.setdefault("position", current_user.position)
+            "is_admin": current_user.is_admin,
-    context_json.setdefault("grade", current_user.grade)
+            "username": current_user.username,
-    context_json.setdefault("employee_no", current_user.employee_no)
+            "name": current_user.name,
-    context_json.setdefault("manager_name", current_user.manager_name)
+            "department_name": current_user.department_name,
+            "position": current_user.position,
+            "grade": current_user.grade,
+            "employee_no": current_user.employee_no,
+            "manager_name": current_user.manager_name,
+        }
+    )
     return context_json
 
 
@@ -134,7 +140,7 @@ def run_application_preview_action(
     run_id = f"application-preview-action:{payload.conversation_id or current_user.username}"
     request = UserAgentRequest(
         run_id=run_id,
-        user_id=payload.user_id or current_user.username or current_user.name,
+        user_id=current_user.username or current_user.name,
         message=payload.message,
         ontology=OntologyParseResult(
             scenario="expense",

server/src/app/schemas/expense_case.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 from datetime import datetime
-from typing import Any
 
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -9,11 +8,30 @@ 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 ArchivedApplicationRead(BaseModel):
+    """付款事件中允许用户看到的关联申请摘要。"""
+
+    model_config = ConfigDict(extra="ignore")
+
+    application_claim_no: str = ""
+
+
+class BusinessEventPayloadRead(BaseModel):
+    """费用时间线允许面向用户展示的事件载荷。"""
+
+    model_config = ConfigDict(extra="ignore")
+
+    previous_status: str = ""
+    next_status: str = ""
+    next_approval_stage: str = ""
+    reason: str = ""
+    opinion: str = ""
+    application_claim_no: str = ""
+    reimbursement_claim_no: str = ""
+    archived_applications: list[ArchivedApplicationRead] = Field(default_factory=list)
 
 
 class BusinessEventRead(BaseModel):
@@ -21,18 +39,10 @@ class BusinessEventRead(BaseModel):
 
     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)
+    payload_json: BusinessEventPayloadRead = Field(default_factory=BusinessEventPayloadRead)
-    delivery_status: str
     occurred_at: datetime
-    published_at: datetime | None
 
 
 class ExpenseCaseTimelineRead(BaseModel):
@@ -40,12 +50,7 @@ class ExpenseCaseTimelineRead(BaseModel):
 
     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/expense_application_draft_events.py

@@ -0,0 +1,124 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import uuid
+from decimal import Decimal
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.api.deps import CurrentUserContext
+from app.models.expense_case import BusinessEvent
+from app.models.financial_record import ExpenseClaim
+from app.schemas.user_agent import UserAgentRequest
+from app.services.expense_cases import ExpenseCaseService
+
+
+class ExpenseApplicationDraftEventService:
+    """为 AI 申请草稿生成稳定、租户隔离的费用事件。"""
+
+    def __init__(self, db: Session) -> None:
+        self.db = db
+
+    def record(
+        self,
+        payload: UserAgentRequest,
+        claim: ExpenseClaim,
+        current_user: CurrentUserContext,
+        *,
+        event_type: str,
+        previous_status: str = "",
+        previous_approval_stage: str = "",
+    ) -> None:
+        ExpenseCaseService(self.db).record_claim_event(
+            claim,
+            event_type=event_type,
+            actor_id=current_user.username,
+            tenant_id=current_user.tenant_id,
+            correlation_id=payload.run_id,
+            idempotency_key=self._build_idempotency_key(
+                payload,
+                claim,
+                current_user,
+                event_type=event_type,
+            ),
+            previous_status=previous_status,
+            previous_approval_stage=previous_approval_stage,
+        )
+
+    def prepare_created_draft(
+        self,
+        payload: UserAgentRequest,
+        claim: ExpenseClaim,
+        current_user: CurrentUserContext,
+    ) -> tuple[str, ExpenseClaim | None]:
+        """生成数据库级稳定聚合 ID，并查找同一动作已创建的草稿。"""
+
+        idempotency_key = self._build_idempotency_key(
+            payload,
+            claim,
+            current_user,
+            event_type="claim_draft_created",
+        )
+        claim.id = str(uuid.uuid5(uuid.NAMESPACE_URL, idempotency_key))
+        existing = self.find_created_draft(current_user, idempotency_key=idempotency_key)
+        if existing is not None:
+            return idempotency_key, existing
+
+        if self.db.get(ExpenseClaim, claim.id) is not None:
+            raise RuntimeError("申请草稿幂等状态不完整，请重试。")
+        return idempotency_key, None
+
+    def find_created_draft(
+        self,
+        current_user: CurrentUserContext,
+        *,
+        idempotency_key: str,
+    ) -> ExpenseClaim | None:
+        event = self.db.scalar(
+            select(BusinessEvent)
+            .where(
+                BusinessEvent.tenant_id == current_user.tenant_id,
+                BusinessEvent.aggregate_type == "expense_claim",
+                BusinessEvent.event_type == "claim_draft_created",
+                BusinessEvent.idempotency_key == idempotency_key,
+                BusinessEvent.actor_id == current_user.username,
+            )
+            .order_by(BusinessEvent.occurred_at.desc())
+            .limit(1)
+        )
+        if event is None:
+            return None
+        return self.db.get(ExpenseClaim, event.aggregate_id)
+
+    @staticmethod
+    def _build_idempotency_key(
+        payload: UserAgentRequest,
+        claim: ExpenseClaim,
+        current_user: CurrentUserContext,
+        *,
+        event_type: str,
+    ) -> str:
+        snapshot = {
+            "tenant_id": str(current_user.tenant_id or "default"),
+            "actor_id": str(current_user.username or "anonymous"),
+            "run_id": str(payload.run_id or "application-draft"),
+            "expense_type": str(claim.expense_type or ""),
+            "reason": str(claim.reason or ""),
+            "location": str(claim.location or ""),
+            "amount": str(claim.amount or Decimal("0.00")),
+            "currency": str(claim.currency or "CNY"),
+            "status": str(claim.status or ""),
+            "approval_stage": str(claim.approval_stage or ""),
+            "risk_flags": claim.risk_flags_json or [],
+        }
+        serialized = json.dumps(
+            snapshot,
+            ensure_ascii=False,
+            sort_keys=True,
+            separators=(",", ":"),
+            default=str,
+        )
+        digest = hashlib.sha256(serialized.encode("utf-8")).hexdigest()
+        return f"application-draft:{event_type}:{digest}"

server/src/app/services/steward_action_executor.py

@@ -422,6 +422,7 @@ class StewardActionExecutor:
             "entry_source": "steward_action_executor",
             "document_type": "expense_application",
             "application_stage": "expense_application",
+            "tenant_id": current_user.tenant_id,
             "role_codes": current_user.role_codes,
             "is_admin": current_user.is_admin,
             "username": current_user.username,

server/src/app/services/user_agent_application.py

@@ -5,6 +5,7 @@ from datetime import UTC, datetime
 from decimal import Decimal, InvalidOperation
 
 from sqlalchemy import or_, select
+from sqlalchemy.exc import IntegrityError
 
 from app.api.deps import CurrentUserContext
 from app.models.financial_record import ExpenseClaim
@@ -24,6 +25,9 @@ from app.services.document_numbering import (
     build_document_number,
     generate_unique_expense_claim_no,
 )
+from app.services.expense_application_draft_events import (
+    ExpenseApplicationDraftEventService,
+)
 from app.services.expense_claim_access_policy import ExpenseClaimAccessPolicy
 from app.services.expense_claim_risk_stage import with_risk_business_stage
 from app.services.travel_reimbursement_calculator import TravelReimbursementCalculatorService
@@ -701,6 +705,8 @@ class UserAgentApplicationPersistenceMixin:
         submit: bool,
     ) -> ExpenseClaim:
         current_user = self._build_application_current_user(payload)
+        previous_status = str(claim.status or "").strip()
+        previous_approval_stage = str(claim.approval_stage or "").strip()
         flags = claim.risk_flags_json
         if isinstance(flags, dict):
             flags = [flags]
@@ -725,9 +731,21 @@ class UserAgentApplicationPersistenceMixin:
             claim.status = "draft"
             claim.approval_stage = "待提交"
             claim.submitted_at = None
-            self.db.commit()
+            try:
-            self.db.refresh(claim)
+                ExpenseApplicationDraftEventService(self.db).record(
-            return claim
+                    payload,
+                    claim,
+                    current_user,
+                    event_type="claim_draft_updated",
+                    previous_status=previous_status,
+                    previous_approval_stage=previous_approval_stage,
+                )
+                self.db.commit()
+                self.db.refresh(claim)
+                return claim
+            except Exception:
+                self.db.rollback()
+                raise
 
         from app.services.expense_claims import ExpenseClaimService
 
@@ -743,15 +761,6 @@ class UserAgentApplicationPersistenceMixin:
         *,
         submit: bool,
     ) -> ExpenseClaim:
-        claim_no = self._build_application_claim_no(payload, facts)
-        existing = self.db.scalar(
-            select(ExpenseClaim)
-            .where(ExpenseClaim.claim_no == claim_no)
-            .limit(1)
-        )
-        if existing is not None:
-            return existing
-
         current_user = self._build_application_current_user(payload)
         access_policy = ExpenseClaimAccessPolicy(self.db)
         employee = access_policy.resolve_current_employee(current_user)
@@ -768,7 +777,7 @@ class UserAgentApplicationPersistenceMixin:
                 department_name = str(employee.organization_unit.name).strip()
 
         claim = ExpenseClaim(
-            claim_no=claim_no,
+            claim_no=self._build_application_claim_no(payload, facts),
             employee_id=employee_id,
             employee_name=employee_name,
             department_id=department_id,
@@ -786,25 +795,52 @@ class UserAgentApplicationPersistenceMixin:
             approval_stage="待提交",
             risk_flags_json=[self._build_application_detail_flag(facts)],
         )
-        self.db.add(claim)
+        draft_event_service = ExpenseApplicationDraftEventService(self.db)
-        self.db.flush()
+        draft_idempotency_key = ""
         if not submit:
-            self.db.commit()
+            draft_idempotency_key, existing = draft_event_service.prepare_created_draft(
-            self.db.refresh(claim)
+                payload,
-            return claim
+                claim,
-
+                current_user,
-        from app.services.expense_claims import ExpenseClaimService
+            )
+            if existing is not None:
+                return existing
 
+        self.db.add(claim)
         try:
+            try:
+                self.db.flush()
+            except IntegrityError:
+                self.db.rollback()
+                if draft_idempotency_key:
+                    existing = draft_event_service.find_created_draft(
+                        current_user,
+                        idempotency_key=draft_idempotency_key,
+                    )
+                    if existing is not None:
+                        return existing
+                raise
+            if not submit:
+                draft_event_service.record(
+                    payload,
+                    claim,
+                    current_user,
+                    event_type="claim_draft_created",
+                )
+                self.db.commit()
+                self.db.refresh(claim)
+                return claim
+
+            from app.services.expense_claims import ExpenseClaimService
+
             submitted = ExpenseClaimService(self.db).submit_claim(claim.id, current_user)
+            if submitted is None:
+                raise ValueError("未找到可提交的申请单。")
+            return submitted
         except Exception:
             # 外层编排会记录失败工具调用并提交事务，必须先回滚本次已 flush 的草稿。
             self.db.rollback()
             raise
-        if submitted is None:
-            self.db.rollback()
-            raise ValueError("未找到可提交的申请单。")
-        return submitted
 
     def _find_duplicate_expense_application_record(
         self,
@@ -974,6 +1010,12 @@ class UserAgentApplicationPersistenceMixin:
             name=name or username or "anonymous",
             role_codes=role_codes,
             is_admin=bool(context_json.get("is_admin")),
+            tenant_id=str(
+                context_json.get("tenant_id")
+                or context_json.get("tenantId")
+                or "default"
+            ).strip()
+            or "default",
             department_name=str(
                 context_json.get("department_name")
                 or context_json.get("department")

server/tests/auth_helpers.py

@@ -23,6 +23,7 @@ def _read_test_user_headers(
     grade: Annotated[str | None, Header(alias="X-Auth-Grade")] = None,
     employee_no: Annotated[str | None, Header(alias="X-Auth-Employee-No")] = None,
     manager_name: Annotated[str | None, Header(alias="X-Auth-Manager-Name")] = None,
+    tenant_id: Annotated[str | None, Header(alias="X-Auth-Tenant-Id")] = None,
 ) -> CurrentUserContext:
     normalized_username = str(username or "").strip()
     normalized_name = str(name or normalized_username).strip()
@@ -44,6 +45,7 @@ def _read_test_user_headers(
         name=normalized_name or normalized_username,
         role_codes=normalized_roles,
         is_admin=admin_flag,
+        tenant_id=str(tenant_id or "default").strip() or "default",
         department_name=str(department or "").strip(),
         cost_center=str(cost_center or "").strip(),
         position=str(position or "").strip(),

server/tests/test_expense_case_endpoints.py

@@ -0,0 +1,339 @@
+from __future__ import annotations
+
+from collections.abc import Generator
+from datetime import UTC, datetime
+from decimal import Decimal
+
+import pytest
+from auth_helpers import install_legacy_header_auth_override
+from fastapi.testclient import TestClient
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.pool import StaticPool
+
+from app.api.deps import get_db
+from app.db.base import Base
+from app.main import create_app
+from app.models.employee import Employee
+from app.models.financial_record import ExpenseClaim
+from app.services.expense_cases import ExpenseCaseService
+from app.services.expense_claim_workflow_constants import (
+    DIRECT_MANAGER_APPROVAL_STAGE,
+    FINANCE_APPROVAL_STAGE,
+)
+
+
+@pytest.fixture
+def http_context() -> Generator[tuple[TestClient, sessionmaker[Session]], None, None]:
+    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)
+    app = create_app()
+    install_legacy_header_auth_override(app)
+
+    def override_db() -> Generator[Session, None, None]:
+        with session_factory() as db:
+            yield db
+
+    app.dependency_overrides[get_db] = override_db
+    client = TestClient(app)
+    try:
+        yield client, session_factory
+    finally:
+        client.close()
+        app.dependency_overrides.clear()
+        engine.dispose()
+
+
+def seed_timeline(
+    db: Session,
+    *,
+    claim_no: str = "RE-TIMELINE-001",
+    expense_type: str = "travel",
+    relation_type: str | None = None,
+    approval_stage: str = DIRECT_MANAGER_APPROVAL_STAGE,
+    with_case: bool = True,
+    tenant_id: str = "default",
+) -> ExpenseClaim:
+    manager = Employee(
+        id="manager-1",
+        employee_no="M001",
+        name="李经理",
+        email="manager@example.com",
+    )
+    owner = Employee(
+        id="owner-1",
+        employee_no="E001",
+        name="张三",
+        email="owner@example.com",
+        manager=manager,
+    )
+    claim = ExpenseClaim(
+        id="claim-1",
+        claim_no=claim_no,
+        employee_id=owner.id,
+        employee_name=owner.name,
+        department_name="市场部",
+        expense_type=expense_type,
+        reason="客户现场差旅",
+        location="上海",
+        amount=Decimal("880.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),
+        status="submitted",
+        approval_stage=approval_stage,
+        risk_flags_json=[],
+    )
+    db.add_all([manager, owner, claim])
+    db.flush()
+
+    if with_case:
+        ExpenseCaseService(db).record_claim_event(
+            claim,
+            event_type="claim_submitted",
+            actor_id=owner.email,
+            tenant_id=tenant_id,
+            correlation_id="timeline-http-permission",
+            idempotency_key="timeline-http-permission",
+            previous_status="draft",
+            previous_approval_stage="待提交",
+            relation_type=relation_type,
+            extra_payload={
+                "opinion": "同意，按计划执行。",
+                "internal_route": "finance-secret-route",
+                "archived_applications": [
+                    {
+                        "application_claim_id": "internal-application-id",
+                        "application_claim_no": "AP-TIMELINE-001",
+                        "archive_event_id": "internal-archive-event-id",
+                    }
+                ],
+            },
+        )
+    db.commit()
+    return claim
+
+
+@pytest.mark.parametrize(
+    ("approval_stage", "headers"),
+    [
+        (
+            DIRECT_MANAGER_APPROVAL_STAGE,
+            {
+                "x-auth-username": "owner@example.com",
+                "x-auth-name": "Owner",
+                "x-auth-employee-no": "E001",
+                "x-auth-role-codes": "user",
+            },
+        ),
+        (
+            DIRECT_MANAGER_APPROVAL_STAGE,
+            {
+                "x-auth-username": "manager@example.com",
+                "x-auth-name": "Manager",
+                "x-auth-employee-no": "M001",
+                "x-auth-role-codes": "approver",
+            },
+        ),
+        (
+            FINANCE_APPROVAL_STAGE,
+            {
+                "x-auth-username": "finance@example.com",
+                "x-auth-name": "Finance",
+                "x-auth-role-codes": "finance",
+            },
+        ),
+        (
+            DIRECT_MANAGER_APPROVAL_STAGE,
+            {
+                "x-auth-username": "admin",
+                "x-auth-name": "Admin",
+                "x-auth-is-admin": "true",
+            },
+        ),
+    ],
+    ids=["owner", "current-manager", "finance", "admin"],
+)
+def test_expense_case_timeline_allows_supported_viewers(
+    http_context: tuple[TestClient, sessionmaker[Session]],
+    approval_stage: str,
+    headers: dict[str, str],
+) -> None:
+    client, session_factory = http_context
+    with session_factory() as db:
+        claim = seed_timeline(db, approval_stage=approval_stage)
+        claim_id = claim.id
+
+    response = client.get(f"/api/v1/expense-cases/by-claim/{claim_id}", headers=headers)
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert set(payload) == {"id", "case_no", "current_stage", "status", "links", "events"}
+    assert payload["case_no"] == "CASE-RE-TIMELINE-001"
+    assert payload["links"] == [{"relation_type": "claim"}]
+    assert len(payload["events"]) == 1
+    event = payload["events"][0]
+    assert set(event) == {
+        "id",
+        "event_type",
+        "actor_id",
+        "actor_type",
+        "payload_json",
+        "occurred_at",
+    }
+    assert event["actor_id"] == "owner@example.com"
+    assert event["payload_json"]["opinion"] == "同意，按计划执行。"
+    assert event["payload_json"]["archived_applications"] == [
+        {"application_claim_no": "AP-TIMELINE-001"}
+    ]
+    serialized = response.text
+    for internal_value in (
+        "idempotency_key",
+        "correlation_id",
+        "causation_id",
+        "delivery_status",
+        "resource_id",
+        "internal_route",
+        "internal-application-id",
+        "internal-archive-event-id",
+    ):
+        assert internal_value not in serialized
+
+
+def test_expense_case_timeline_hides_claim_existence_from_unrelated_user(
+    http_context: tuple[TestClient, sessionmaker[Session]],
+) -> None:
+    client, session_factory = http_context
+    with session_factory() as db:
+        claim_id = seed_timeline(db).id
+
+    response = client.get(
+        f"/api/v1/expense-cases/by-claim/{claim_id}",
+        headers={
+            "x-auth-username": "unrelated@example.com",
+            "x-auth-name": "Unrelated User",
+            "x-auth-employee-no": "E999",
+            "x-auth-role-codes": "user",
+        },
+    )
+
+    assert response.status_code == 404
+    assert response.json()["detail"] == "费用单据不存在。"
+
+
+def test_expense_case_timeline_rejects_cross_tenant_lookup(
+    http_context: tuple[TestClient, sessionmaker[Session]],
+) -> None:
+    client, session_factory = http_context
+    with session_factory() as db:
+        claim_id = seed_timeline(db, tenant_id="tenant-a").id
+
+    response = client.get(
+        f"/api/v1/expense-cases/by-claim/{claim_id}",
+        headers={
+            "x-auth-username": "owner@example.com",
+            "x-auth-name": "Owner",
+            "x-auth-employee-no": "E001",
+            "x-auth-role-codes": "user",
+            "x-auth-tenant-id": "tenant-b",
+        },
+    )
+
+    assert response.status_code == 404
+    assert response.json()["detail"] == "该单据尚未纳入统一费用事件。"
+
+
+def test_expense_case_timeline_returns_not_covered_for_claim_without_case(
+    http_context: tuple[TestClient, sessionmaker[Session]],
+) -> None:
+    client, session_factory = http_context
+    with session_factory() as db:
+        claim_id = seed_timeline(db, with_case=False).id
+
+    response = client.get(
+        f"/api/v1/expense-cases/by-claim/{claim_id}",
+        headers={
+            "x-auth-username": "owner@example.com",
+            "x-auth-name": "Owner",
+            "x-auth-employee-no": "E001",
+            "x-auth-role-codes": "user",
+        },
+    )
+
+    assert response.status_code == 404
+    assert response.json()["detail"] == "该单据尚未纳入统一费用事件。"
+
+
+def test_expense_case_timeline_returns_safe_summary_for_all_linked_claims(
+    http_context: tuple[TestClient, sessionmaker[Session]],
+) -> None:
+    client, session_factory = http_context
+    with session_factory() as db:
+        application = seed_timeline(
+            db,
+            claim_no="AP-TIMELINE-001",
+            expense_type="travel_application",
+            relation_type="application",
+        )
+        service = ExpenseCaseService(db)
+        expense_case = service.get_timeline_for_claim(application.id)
+        assert expense_case is not None
+        reimbursement = ExpenseClaim(
+            id="claim-2",
+            claim_no="RE-TIMELINE-002",
+            employee_id=application.employee_id,
+            employee_name=application.employee_name,
+            department_name=application.department_name,
+            expense_type="travel",
+            reason="客户现场差旅报销",
+            location="上海",
+            amount=Decimal("880.00"),
+            currency="CNY",
+            invoice_count=1,
+            occurred_at=application.occurred_at,
+            submitted_at=application.submitted_at,
+            status="submitted",
+            approval_stage=FINANCE_APPROVAL_STAGE,
+            risk_flags_json=[],
+        )
+        db.add(reimbursement)
+        db.flush()
+        service.record_claim_event(
+            reimbursement,
+            event_type="claim_submitted",
+            actor_id="owner@example.com",
+            expense_case=expense_case,
+            relation_type="generated_reimbursement",
+            correlation_id="linked-reimbursement",
+            idempotency_key="linked-reimbursement",
+            previous_status="draft",
+            previous_approval_stage="待提交",
+        )
+        db.commit()
+
+    response = client.get(
+        "/api/v1/expense-cases/by-claim/claim-1",
+        headers={
+            "x-auth-username": "owner@example.com",
+            "x-auth-name": "Owner",
+            "x-auth-employee-no": "E001",
+            "x-auth-role-codes": "user",
+        },
+    )
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert {item["relation_type"] for item in payload["links"]} == {
+        "application",
+        "generated_reimbursement",
+    }
+    assert [event["event_type"] for event in payload["events"]] == [
+        "claim_submitted",
+        "claim_submitted",
+    ]

server/tests/test_reimbursement_endpoints.py

@@ -1114,9 +1114,147 @@ def test_application_preview_action_saves_draft_with_detail_reference(monkeypatc
         assert db.scalar(
             select(BudgetReservation).where(BudgetReservation.source_id == claim.id)
         ) is None
-        assert db.scalar(
+        event = db.scalar(
             select(BusinessEvent).where(BusinessEvent.aggregate_id == claim.id)
-        ) is None
+        )
-        assert db.scalar(
+        assert event is not None
+        assert event.event_type == "claim_draft_created"
+        assert event.payload_json["previous_status"] == ""
+        assert event.payload_json["next_status"] == "draft"
+        link = db.scalar(
             select(ExpenseCaseLink).where(ExpenseCaseLink.resource_id == claim.id)
+        )
+        assert link is not None
+        assert link.relation_type == "application"
+
+
+def test_application_preview_action_rejects_forged_identity_when_editing_other_claim() -> None:
+    client, session_factory = build_client()
+    with session_factory() as db:
+        seed_claim(db)
+        outsider = Employee(
+            id="emp-outsider",
+            employee_no="E90001",
+            name="其他员工",
+            email="outsider@example.com",
+        )
+        outsider_claim = ExpenseClaim(
+            id="application-outsider-1",
+            claim_no="AP-OUTSIDER-001",
+            employee_id=outsider.id,
+            employee_name=outsider.name,
+            department_name="交付部",
+            expense_type="travel_application",
+            reason="其他员工原申请",
+            location="北京",
+            amount=Decimal("500.00"),
+            currency="CNY",
+            invoice_count=0,
+            occurred_at=datetime(2026, 7, 10, tzinfo=UTC),
+            submitted_at=None,
+            status="returned",
+            approval_stage="退回补充",
+            risk_flags_json=[],
+        )
+        db.add_all([outsider, outsider_claim])
+        db.commit()
+
+    response = client.post(
+        "/api/v1/reimbursements/application-preview-action",
+        headers={
+            "x-auth-username": "zhangsan@example.com",
+            "x-auth-name": "Zhang San",
+            "x-auth-employee-no": "E10001",
+            "x-auth-role-codes": "user",
+        },
+        json={
+            "source": "user_message",
+            "user_id": "outsider@example.com",
+            "conversation_id": "conversation-forged-identity",
+            "message": "费用申请保存草稿\n申请时间：2026-07-13 至 2026-07-14\n地点：上海\n事由：恶意修改\n申请金额：880元\n保存草稿",
+            "context_json": {
+                "session_type": "application",
+                "application_action": "save_draft",
+                "application_save_mode": True,
+                "application_edit_mode": True,
+                "application_edit_claim_id": "application-outsider-1",
+                "username": "outsider@example.com",
+                "name": "其他员工",
+                "employee_no": "E90001",
+                "role_codes": ["admin"],
+                "is_admin": True,
+            },
+        },
+    )
+
+    assert response.status_code == 400
+    assert response.json()["detail"] == "只能修改本人被退回的申请单。"
+    with session_factory() as db:
+        persisted = db.get(ExpenseClaim, "application-outsider-1")
+        assert persisted is not None
+        assert persisted.reason == "其他员工原申请"
+        assert persisted.status == "returned"
+        assert persisted.approval_stage == "退回补充"
+        assert db.scalar(
+            select(BusinessEvent).where(BusinessEvent.aggregate_id == persisted.id)
         ) is None
+
+
+def test_application_preview_action_reuses_created_draft_for_identical_retry() -> None:
+    client, session_factory = build_client()
+    with session_factory() as db:
+        seed_claim(db)
+
+    request_payload = {
+        "source": "user_message",
+        "user_id": "zhangsan@example.com",
+        "conversation_id": "conversation-fast-save-retry",
+        "message": "费用申请保存草稿\n地点：上海\n事由：项目验收\n申请金额：880元\n保存草稿",
+        "context_json": {
+            "session_type": "application",
+            "application_action": "save_draft",
+            "application_save_mode": True,
+        },
+    }
+    headers = {
+        "x-auth-username": "zhangsan@example.com",
+        "x-auth-name": "Zhang San",
+        "x-auth-employee-no": "E10001",
+        "x-auth-role-codes": "user",
+    }
+
+    first_response = client.post(
+        "/api/v1/reimbursements/application-preview-action",
+        headers=headers,
+        json=request_payload,
+    )
+    second_response = client.post(
+        "/api/v1/reimbursements/application-preview-action",
+        headers=headers,
+        json=request_payload,
+    )
+
+    assert first_response.status_code == 200
+    assert second_response.status_code == 200
+    first_draft = first_response.json()["result"]["draft_payload"]
+    second_draft = second_response.json()["result"]["draft_payload"]
+    assert second_draft["claim_id"] == first_draft["claim_id"]
+    assert second_draft["claim_no"] == first_draft["claim_no"]
+    with session_factory() as db:
+        application_claims = list(
+            db.scalars(
+                select(ExpenseClaim).where(
+                    ExpenseClaim.expense_type == "travel_application"
+                )
+            ).all()
+        )
+        assert len(application_claims) == 1
+        events = list(
+            db.scalars(
+                select(BusinessEvent).where(
+                    BusinessEvent.aggregate_id == application_claims[0].id
+                )
+            ).all()
+        )
+        assert len(events) == 1
+        assert events[0].event_type == "claim_draft_created"

server/tests/test_steward_action_executor.py

@@ -14,6 +14,7 @@ from app.db.base import Base
 from app.main import create_app
 from app.models.agent_conversation import AgentConversation
 from app.models.employee import Employee
+from app.models.expense_case import BusinessEvent, ExpenseCaseLink
 from app.models.financial_record import ExpenseClaim
 from app.services import attachment_association_jobs as attachment_jobs_module
 
@@ -284,6 +285,9 @@ def test_steward_action_executor_reuses_checkpoint_for_duplicate_trace_without_d
     assert second_payload["result_payload"]["idempotent_replay"] is True
     with session_factory() as db:
         assert claim_count(db) == 1
+        events = list(db.scalars(select(BusinessEvent)).all())
+        assert len(events) == 1
+        assert events[0].event_type == "claim_draft_created"
 
 
 def test_steward_action_executor_requires_confirmation_before_submit_side_effect() -> None:
@@ -342,6 +346,16 @@ def test_steward_action_executor_saves_application_draft_from_action_step() -> N
         claim = db.scalars(select(ExpenseClaim)).one()
         assert claim.status == "draft"
         assert claim.reason == "辅助国网仿生产服务器部署"
+        event = db.scalars(
+            select(BusinessEvent).where(BusinessEvent.aggregate_id == claim.id)
+        ).one()
+        assert event.event_type == "claim_draft_created"
+        assert event.actor_id == "zhangsan@example.com"
+        assert event.correlation_id == "steward-action:save_application_draft:task_app_001"
+        link = db.scalars(
+            select(ExpenseCaseLink).where(ExpenseCaseLink.resource_id == claim.id)
+        ).one()
+        assert link.relation_type == "application"
 
 
 def test_steward_action_executor_creates_reimbursement_draft_from_action_step() -> None:

server/tests/test_user_agent_application_draft_events.py

@@ -0,0 +1,285 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from decimal import Decimal
+
+import pytest
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.pool import StaticPool
+
+from app.db.base import Base
+from app.models.employee import Employee
+from app.models.expense_case import BusinessEvent, ExpenseCaseLink
+from app.models.financial_record import ExpenseClaim
+from app.schemas.ontology import OntologyParseResult
+from app.schemas.user_agent import UserAgentRequest
+from app.services.expense_cases import ExpenseCaseService
+from app.services.user_agent import UserAgentService
+
+
+def build_session_factory() -> sessionmaker[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_request(*, run_id: str, tenant_id: str = "tenant-a") -> UserAgentRequest:
+    return UserAgentRequest(
+        run_id=run_id,
+        user_id="owner@example.com",
+        message="保存申请草稿",
+        ontology=OntologyParseResult(run_id=run_id),
+        context_json={
+            "session_type": "application",
+            "tenant_id": tenant_id,
+            "name": "张三",
+            "employee_no": "E001",
+            "department_name": "市场部",
+        },
+    )
+
+
+def build_facts(*, reason: str = "客户现场差旅") -> dict[str, str]:
+    return {
+        "application_type": "差旅费用申请",
+        "time": "2026-07-13 至 2026-07-14",
+        "location": "上海",
+        "reason": reason,
+        "amount": "880元",
+    }
+
+
+def test_ai_application_draft_update_writes_tenant_scoped_event() -> None:
+    session_factory = build_session_factory()
+    with session_factory() as db:
+        owner = Employee(
+            id="owner-1",
+            employee_no="E001",
+            name="张三",
+            email="owner@example.com",
+        )
+        claim = ExpenseClaim(
+            id="application-1",
+            claim_no="AP-DRAFT-001",
+            employee_id=owner.id,
+            employee_name=owner.name,
+            department_name="市场部",
+            expense_type="travel_application",
+            reason="原申请事由",
+            location="北京",
+            amount=Decimal("500.00"),
+            currency="CNY",
+            invoice_count=0,
+            occurred_at=datetime(2026, 7, 10, tzinfo=UTC),
+            submitted_at=None,
+            status="returned",
+            approval_stage="退回补充",
+            risk_flags_json=[],
+        )
+        db.add_all([owner, claim])
+        db.commit()
+
+        updated = UserAgentService(db)._update_expense_application_record(
+            build_request(run_id="application-draft-update"),
+            build_facts(reason="更新后的申请事由"),
+            claim,
+            submit=False,
+        )
+
+        assert updated.status == "draft"
+        event = db.scalar(select(BusinessEvent).where(BusinessEvent.aggregate_id == claim.id))
+        assert event is not None
+        assert event.event_type == "claim_draft_updated"
+        assert event.tenant_id == "tenant-a"
+        assert event.actor_id == "owner@example.com"
+        assert event.correlation_id == "application-draft-update"
+        assert event.payload_json["previous_status"] == "returned"
+        assert event.payload_json["next_status"] == "draft"
+        link = db.scalar(select(ExpenseCaseLink).where(ExpenseCaseLink.resource_id == claim.id))
+        assert link is not None
+        assert link.tenant_id == "tenant-a"
+
+
+def test_ai_application_draft_creation_rolls_back_when_event_write_fails(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    session_factory = build_session_factory()
+
+    def fail_event(*_args, **_kwargs):
+        raise RuntimeError("模拟草稿事件写入失败")
+
+    monkeypatch.setattr(ExpenseCaseService, "record_claim_event", fail_event)
+    with session_factory() as db:
+        with pytest.raises(RuntimeError, match="模拟草稿事件写入失败"):
+            UserAgentService(db)._create_expense_application_record(
+                build_request(run_id="application-draft-rollback"),
+                build_facts(),
+                submit=False,
+            )
+
+        db.commit()
+        assert list(db.scalars(select(ExpenseClaim)).all()) == []
+        assert list(db.scalars(select(ExpenseCaseLink)).all()) == []
+        assert list(db.scalars(select(BusinessEvent)).all()) == []
+
+
+def test_ai_application_draft_update_rolls_back_when_event_write_fails(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    session_factory = build_session_factory()
+    with session_factory() as db:
+        owner = Employee(
+            id="owner-1",
+            employee_no="E001",
+            name="张三",
+            email="owner@example.com",
+        )
+        claim = ExpenseClaim(
+            id="application-1",
+            claim_no="AP-DRAFT-ROLLBACK-001",
+            employee_id=owner.id,
+            employee_name=owner.name,
+            department_name="市场部",
+            expense_type="travel_application",
+            reason="原申请事由",
+            location="北京",
+            amount=Decimal("500.00"),
+            currency="CNY",
+            invoice_count=0,
+            occurred_at=datetime(2026, 7, 10, tzinfo=UTC),
+            submitted_at=None,
+            status="returned",
+            approval_stage="退回补充",
+            risk_flags_json=[],
+        )
+        db.add_all([owner, claim])
+        db.commit()
+
+        def fail_event(*_args, **_kwargs):
+            raise RuntimeError("模拟草稿事件写入失败")
+
+        monkeypatch.setattr(ExpenseCaseService, "record_claim_event", fail_event)
+        with pytest.raises(RuntimeError, match="模拟草稿事件写入失败"):
+            UserAgentService(db)._update_expense_application_record(
+                build_request(run_id="application-draft-update-rollback"),
+                build_facts(reason="不应落库的新事由"),
+                claim,
+                submit=False,
+            )
+
+        db.commit()
+        persisted = db.get(ExpenseClaim, claim.id)
+        assert persisted is not None
+        assert persisted.reason == "原申请事由"
+        assert persisted.location == "北京"
+        assert persisted.amount == Decimal("500.00")
+        assert persisted.status == "returned"
+        assert persisted.approval_stage == "退回补充"
+        assert list(db.scalars(select(ExpenseCaseLink)).all()) == []
+        assert list(db.scalars(select(BusinessEvent)).all()) == []
+
+
+def test_ai_application_draft_update_deduplicates_identical_snapshot() -> None:
+    session_factory = build_session_factory()
+    with session_factory() as db:
+        owner = Employee(
+            id="owner-1",
+            employee_no="E001",
+            name="张三",
+            email="owner@example.com",
+        )
+        claim = ExpenseClaim(
+            id="application-1",
+            claim_no="AP-DRAFT-IDEMPOTENT-001",
+            employee_id=owner.id,
+            employee_name=owner.name,
+            department_name="市场部",
+            expense_type="travel_application",
+            reason="原申请事由",
+            location="北京",
+            amount=Decimal("500.00"),
+            currency="CNY",
+            invoice_count=0,
+            occurred_at=datetime(2026, 7, 10, tzinfo=UTC),
+            submitted_at=None,
+            status="returned",
+            approval_stage="退回补充",
+            risk_flags_json=[],
+        )
+        db.add_all([owner, claim])
+        db.commit()
+
+        request = build_request(run_id="application-draft-idempotent")
+        facts = build_facts(reason="同一份草稿")
+        service = UserAgentService(db)
+        service._update_expense_application_record(request, facts, claim, submit=False)
+        service._update_expense_application_record(request, facts, claim, submit=False)
+
+        events = list(
+            db.scalars(
+                select(BusinessEvent).where(BusinessEvent.aggregate_id == claim.id)
+            ).all()
+        )
+        assert len(events) == 1
+        assert events[0].event_type == "claim_draft_updated"
+        assert events[0].payload_json["previous_status"] == "returned"
+
+
+def test_ai_application_draft_update_keeps_distinct_snapshots_in_same_run() -> None:
+    session_factory = build_session_factory()
+    with session_factory() as db:
+        owner = Employee(
+            id="owner-1",
+            employee_no="E001",
+            name="张三",
+            email="owner@example.com",
+        )
+        claim = ExpenseClaim(
+            id="application-1",
+            claim_no="AP-DRAFT-VERSIONS-001",
+            employee_id=owner.id,
+            employee_name=owner.name,
+            department_name="市场部",
+            expense_type="travel_application",
+            reason="原申请事由",
+            location="北京",
+            amount=Decimal("500.00"),
+            currency="CNY",
+            invoice_count=0,
+            occurred_at=datetime(2026, 7, 10, tzinfo=UTC),
+            submitted_at=None,
+            status="returned",
+            approval_stage="退回补充",
+            risk_flags_json=[],
+        )
+        db.add_all([owner, claim])
+        db.commit()
+
+        request = build_request(run_id="application-draft-multi-version")
+        service = UserAgentService(db)
+        service._update_expense_application_record(
+            request,
+            build_facts(reason="第一版草稿"),
+            claim,
+            submit=False,
+        )
+        service._update_expense_application_record(
+            request,
+            build_facts(reason="第二版草稿"),
+            claim,
+            submit=False,
+        )
+
+        events = list(
+            db.scalars(
+                select(BusinessEvent).where(BusinessEvent.aggregate_id == claim.id)
+            ).all()
+        )
+        assert len(events) == 2
+        assert {event.event_type for event in events} == {"claim_draft_updated"}
+        assert len({event.idempotency_key for event in events}) == 2