From 1347366b95f24e5b187f24a57b4a335c048b55be Mon Sep 17 00:00:00 2001 From: caoxiaozhu Date: Tue, 14 Jul 2026 00:07:07 +0800 Subject: [PATCH] feat(expenses): secure timeline and draft events --- .../application-draft-replay-duplicates.md | 8 + .../application-preview-forged-identity.md | 8 + .../CONCEPT.md | 7 +- .../TODO.md | 12 +- .../app/api/v1/endpoints/reimbursements.py | 26 +- server/src/app/schemas/expense_case.py | 43 ++- .../expense_application_draft_events.py | 124 +++++++ .../app/services/steward_action_executor.py | 1 + .../app/services/user_agent_application.py | 90 +++-- server/tests/auth_helpers.py | 2 + server/tests/test_expense_case_endpoints.py | 339 ++++++++++++++++++ server/tests/test_reimbursement_endpoints.py | 144 +++++++- server/tests/test_steward_action_executor.py | 14 + ...est_user_agent_application_draft_events.py | 285 +++++++++++++++ 14 files changed, 1044 insertions(+), 59 deletions(-) create mode 100644 document/development/2026-07-13/dev-logs/bugs/application-draft-replay-duplicates.md create mode 100644 document/development/2026-07-13/dev-logs/bugs/application-preview-forged-identity.md create mode 100644 server/src/app/services/expense_application_draft_events.py create mode 100644 server/tests/test_expense_case_endpoints.py create mode 100644 server/tests/test_user_agent_application_draft_events.py diff --git a/document/development/2026-07-13/dev-logs/bugs/application-draft-replay-duplicates.md b/document/development/2026-07-13/dev-logs/bugs/application-draft-replay-duplicates.md new file mode 100644 index 0000000..c2177a8 --- /dev/null +++ b/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 双会话并发集成测试仍是后续可补的非阻断验证，不影响当前顺序重放与事务契约。 diff --git a/document/development/2026-07-13/dev-logs/bugs/application-preview-forged-identity.md b/document/development/2026-07-13/dev-logs/bugs/application-preview-forged-identity.md new file mode 100644 index 0000000..3fd24ae --- /dev/null +++ b/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`、角色或员工编号绕过申请所有权检查，授权事实与其他受保护接口统一来自服务端会话。 diff --git a/document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/CONCEPT.md b/document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/CONCEPT.md index 3bc9174..4301a60 100644 --- a/document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/CONCEPT.md +++ b/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` 表，浏览器登录无法获得有效认证凭证，因此本轮未声称完成真实页面端到端联调，也未擅自执行数据库迁移。计划、消费/票据、入账、对账和复盘事件仍待后续补齐。 diff --git a/document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/TODO.md b/document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/TODO.md index 8cca35e..6d9d901 100644 --- a/document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/TODO.md +++ b/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、构建、端到端和未覆盖风险证据。 diff --git a/server/src/app/api/v1/endpoints/reimbursements.py b/server/src/app/api/v1/endpoints/reimbursements.py index 968271d..0ebe1ff 100644 --- a/server/src/app/api/v1/endpoints/reimbursements.py +++ b/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.setdefault("username", current_user.username) - context_json.setdefault("name", current_user.name) - context_json.setdefault("department_name", current_user.department_name) - context_json.setdefault("position", current_user.position) - context_json.setdefault("grade", current_user.grade) - context_json.setdefault("employee_no", current_user.employee_no) - context_json.setdefault("manager_name", current_user.manager_name) + # 身份与权限字段只能来自服务端会话，不能保留请求体中的同名值。 + context_json.update( + { + "tenant_id": current_user.tenant_id, + "role_codes": current_user.role_codes, + "is_admin": current_user.is_admin, + "username": current_user.username, + "name": current_user.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", diff --git a/server/src/app/schemas/expense_case.py b/server/src/app/schemas/expense_case.py index bf48d21..9e335f3 100644 --- a/server/src/app/schemas/expense_case.py +++ b/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) - delivery_status: str + payload_json: BusinessEventPayloadRead = Field(default_factory=BusinessEventPayloadRead) 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) diff --git a/server/src/app/services/expense_application_draft_events.py b/server/src/app/services/expense_application_draft_events.py new file mode 100644 index 0000000..b8b5a7a --- /dev/null +++ b/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}" diff --git a/server/src/app/services/steward_action_executor.py b/server/src/app/services/steward_action_executor.py index 3b802ef..ccc782f 100644 --- a/server/src/app/services/steward_action_executor.py +++ b/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, diff --git a/server/src/app/services/user_agent_application.py b/server/src/app/services/user_agent_application.py index a28f4d1..2d5c327 100644 --- a/server/src/app/services/user_agent_application.py +++ b/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() - self.db.refresh(claim) - return claim + try: + ExpenseApplicationDraftEventService(self.db).record( + 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) - self.db.flush() + draft_event_service = ExpenseApplicationDraftEventService(self.db) + draft_idempotency_key = "" if not submit: - self.db.commit() - self.db.refresh(claim) - return claim - - from app.services.expense_claims import ExpenseClaimService + draft_idempotency_key, existing = draft_event_service.prepare_created_draft( + payload, + claim, + current_user, + ) + 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") diff --git a/server/tests/auth_helpers.py b/server/tests/auth_helpers.py index 902017e..c320275 100644 --- a/server/tests/auth_helpers.py +++ b/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(), diff --git a/server/tests/test_expense_case_endpoints.py b/server/tests/test_expense_case_endpoints.py new file mode 100644 index 0000000..4d07057 --- /dev/null +++ b/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", + ] diff --git a/server/tests/test_reimbursement_endpoints.py b/server/tests/test_reimbursement_endpoints.py index 302803c..4f9c890 100644 --- a/server/tests/test_reimbursement_endpoints.py +++ b/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" diff --git a/server/tests/test_steward_action_executor.py b/server/tests/test_steward_action_executor.py index 1bc57f9..7b2b22e 100644 --- a/server/tests/test_steward_action_executor.py +++ b/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: diff --git a/server/tests/test_user_agent_application_draft_events.py b/server/tests/test_user_agent_application_draft_events.py new file mode 100644 index 0000000..678314f --- /dev/null +++ b/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