diff --git a/document/development/2026-07-13/dev-logs/bugs/ai-application-submit-bypasses-transaction.md b/document/development/2026-07-13/dev-logs/bugs/ai-application-submit-bypasses-transaction.md new file mode 100644 index 0000000..2914404 --- /dev/null +++ b/document/development/2026-07-13/dev-logs/bugs/ai-application-submit-bypasses-transaction.md @@ -0,0 +1,9 @@ +## 修复记录 + +- 15:33：记录 bug 修复：AI 新建费用申请直接提交绕过统一事务。 + - Git 提交检查：`git fetch --all --prune` 后未发现 upstream 新提交；本地 ahead 2 个既有提交，分别为 `653eda05 feat(auth): add opaque bearer sessions`（不透明 Bearer 会话与认证收口）和 `661990b2 feat(expenses): add transactional expense case events`（Expense Case 与事务业务事件基础）。 + - 修改：`user_agent_application.py` 将 AI 新建并直接提交改为先创建草稿，再统一调用 `ExpenseClaimService.submit_claim`，删除入口内手写提交状态和平台风险评估；提交异常或未找到单据时主动回滚，避免外层工具日志提交失败草稿。 + - 修改：`budget.py` 的预算就绪检查改为复用当前 Session 连接执行 metadata 检查，避免 Engine 连接隐式提交已经 `flush` 的申请和预算数据；`test_reimbursement_endpoints.py`、`test_user_agent_service.py` 补齐成功提交、事件失败整体回滚、保存草稿无副作用和必填部门上下文回归。 + - 操作：所有后端验证均在 `x-financial-local-linux` 容器内执行，单条命令使用 `timeout 60s`；没有修改数据库结构，没有对持久化开发数据库执行迁移，也没有重启服务。 + - 验证：AI 直接提交/失败回滚/保存草稿 3 项、申请提交主链路 5 项、预算与 Expense Case 13 项，共 21 项通过；相关 Python 文件 `ruff --select F,I` 通过。整份 `test_reimbursement_endpoints.py` 运行结果为 13 项通过、3 项未通过，失败分别位于既有附件风险等级断言、申请审批路由断言和中文测试请求头编码，不属于本次 AI 提交事务断言。 + - 影响：AI 一键新建申请现在与编辑重提共用同一提交语义，预算预占、提交校验、申请风险标记、Expense Case 事件和审批状态保持一致；业务事件写入失败时，申请、预算额度、预算流水、预算预占和 Case 关联不会残留部分成功数据。 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 21323fd..b149a72 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 @@ -488,3 +488,6 @@ docker exec -w /app -e SERVER_VENV_DIR=/tmp/x-financial-server-venv \ - 2026-07-13（管理面收口）：移除生产代码中的 `X-Auth-*` 授权来源，保护 Settings、模型连通性、缓存、员工、分析、Agent 运行/轨迹、风险观测、审计及系统日志；已初始化 Bootstrap 返回脱敏状态并拒绝匿名重配置，Vite Setup 桥同步锁定。 - 2026-07-13（前端会话）：Web 请求、流式响应和页面关闭收尾统一使用 Bearer，token 与过期时间只保存在 `sessionStorage`，集中处理 `401`、空闲过期和服务端过期；业务经理不再被前端视为平台管理员。 - 2026-07-13（认证验证）：容器内认证/Bootstrap/费用事件/OpenAPI 定向测试 21 项通过，受保护业务端点回归 13 项通过，全量测试收集 805 项成功；前端会话、请求、Setup 锁和权限测试 17 项通过，生产构建通过。迁移仅生成并检查 upgrade/downgrade 离线 SQL，未写入持久化开发数据库。 +- 2026-07-13（P0 提交一致性补缝）：AI 新建费用申请并直接提交时，先持久化为草稿，再统一委托 `ExpenseClaimService.submit_claim`；提交校验、预算预占、申请提交风险标记、`application_submitted` 事件和审批状态不再由 AI 入口分别维护。 +- 2026-07-13（事务边界修复）：预算表运行时就绪检查改为复用当前 Session 连接，避免通过 Engine 执行 metadata 检查时隐式提交已 `flush` 的申请；即使预算预占后 Expense Case 事件写入失败，申请、预算额度、预算流水、预占和 Case 数据也会整体回滚。 +- 2026-07-13（提交一致性验证）：容器内 AI 申请直接提交、失败回滚、保存草稿副作用、申请提交主链路及预算/Expense Case 回归共 21 项通过，`ruff --select F,I` 通过；未修改数据库结构，未对持久化开发数据库执行迁移。 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 e3cdb71..e9dd9c0 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 @@ -64,6 +64,8 @@ - [ ] [CONCEPT: 业务事件与 AI 决策] 建立事务 Outbox：申请、提交、退回、审批、支付和入账状态与事件同事务提交，消费端按事件 ID 幂等处理。 - [x] [CONCEPT: 业务事件与 AI 决策] 完成首批草稿、提交、退回、审批、申请转报销、付款和申请归档事件旁写，具备 correlation、causation、幂等键及事务回滚。 证据：`expense_claim_draft_flow.py`、`expense_claims.py`、`expense_claim_approval_flow.py`、`test_expense_case_service.py`；容器测试覆盖同 Case 关联、重复事件去重及 Outbox 失败整体回滚。 +- [x] [CONCEPT: 业务事件与 AI 决策] 收口 AI 新建申请直接提交入口，统一复用申请提交事务，并消除预算 metadata 检查对外层事务的隐式提交。 + 证据：`user_agent_application.py`、`budget.py`、`test_reimbursement_endpoints.py`、`test_user_agent_service.py`；容器测试覆盖有效预算预占、`application_submitted` 事件、提交风险标记、事件失败整体回滚及保存草稿无提交副作用。 - [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` 映射；切换后禁止关键事件可丢弃写入。 @@ -136,6 +138,8 @@ - [x] [CONCEPT: 测试方案] 完成 Expense Case/Link、业务事件幂等、租户边界、同事务回滚、申请转报销同 Case 和付款归档事件首批测试。 证据：容器内 `pytest -q server/tests/test_expense_case_service.py` 7 项通过；联合差旅主链路定向回归共 24 项通过。 +- [x] [CONCEPT: 测试方案] 为 AI 新建申请直接提交补充统一事务回归，覆盖提交成功、事件失败回滚和仅保存草稿三条边界。 + 证据：容器内直接提交定向测试 3 项、申请提交回归 5 项、预算与 Expense Case 回归 13 项通过；相关 Python 文件 `ruff --select F,I` 通过。 - [ ] [CONCEPT: 测试方案] 为 Expense Case 状态机、事件账本、AI 决策、结果、记忆、自动化和节省服务补充单元测试。 - [x] [CONCEPT: 测试方案] 为服务端会话、管理员保护和 Bootstrap 重配置补充首批安全回归测试。 证据：`test_auth_session_endpoints.py`、`test_auth_service.py`、`test_bootstrap_security.py`；容器定向测试覆盖 token 摘要、伪造身份头、过期/撤销、登出原子收尾、业务经理越权和初始化后匿名重配置拒绝。 diff --git a/server/src/app/services/budget.py b/server/src/app/services/budget.py index 2af453a..9284b6e 100644 --- a/server/src/app/services/budget.py +++ b/server/src/app/services/budget.py @@ -15,8 +15,8 @@ from app.schemas.budget import ( BudgetAllocationRead, BudgetCheckRead, BudgetCheckRequest, - BudgetOperationRequest, BudgetOperationRead, + BudgetOperationRequest, BudgetSummaryRead, BudgetTransactionRead, ) @@ -31,7 +31,8 @@ class BudgetService(BudgetPaginationMixin, BudgetSupportMixin): self.db = db def ensure_budget_ready(self) -> None: - Base.metadata.create_all(bind=self.db.get_bind()) + # 复用当前 Session 连接，避免在业务事务中通过 Engine 隐式提交已 flush 的单据。 + Base.metadata.create_all(bind=self.db.connection()) exists = self.db.scalar(select(BudgetAllocation.id).limit(1)) if exists: return diff --git a/server/src/app/services/user_agent_application.py b/server/src/app/services/user_agent_application.py index 0f3b611..a28f4d1 100644 --- a/server/src/app/services/user_agent_application.py +++ b/server/src/app/services/user_agent_application.py @@ -781,26 +781,30 @@ class UserAgentApplicationPersistenceMixin: currency="CNY", invoice_count=0, occurred_at=self._parse_application_occurred_at(facts.get("time", "")), - submitted_at=datetime.now(UTC) if submit else None, - status="submitted" if submit else "draft", - approval_stage="直属领导审批" if submit else "待提交", + submitted_at=None, + status="draft", + approval_stage="待提交", risk_flags_json=[self._build_application_detail_flag(facts)], ) self.db.add(claim) self.db.flush() - if submit: - from app.services.expense_claims import ExpenseClaimService + if not submit: + self.db.commit() + self.db.refresh(claim) + return claim - platform_review = ExpenseClaimService(self.db).evaluate_platform_risk_rules( - claim, - business_stage="expense_application", - ) - platform_flags = list(platform_review.get("flags") or []) - if platform_flags: - claim.risk_flags_json = [*list(claim.risk_flags_json or []), *platform_flags] - self.db.commit() - self.db.refresh(claim) - return claim + from app.services.expense_claims import ExpenseClaimService + + try: + submitted = ExpenseClaimService(self.db).submit_claim(claim.id, current_user) + 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, diff --git a/server/tests/test_reimbursement_endpoints.py b/server/tests/test_reimbursement_endpoints.py index 68078f2..302803c 100644 --- a/server/tests/test_reimbursement_endpoints.py +++ b/server/tests/test_reimbursement_endpoints.py @@ -6,24 +6,31 @@ from collections.abc import Generator from datetime import UTC, date, 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 import create_engine, select 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.budget import BudgetAllocation, BudgetReservation, BudgetTransaction from app.models.employee import Employee +from app.models.expense_case import BusinessEvent, ExpenseCaseLink from app.models.financial_record import ExpenseClaim, ExpenseClaimItem from app.models.organization import OrganizationUnit from app.models.risk_observation import RiskObservation, RiskObservationFeedback from app.models.role import Role from app.schemas.ocr import OcrRecognizeBatchRead, OcrRecognizeDocumentRead +from app.schemas.ontology import OntologyParseResult +from app.schemas.user_agent import UserAgentRequest from app.services.document_preview import DocumentPreviewAssets +from app.services.expense_cases import ExpenseCaseService from app.services.expense_claim_attachment_storage import ExpenseClaimAttachmentStorage from app.services.ocr import OcrService +from app.services.user_agent import UserAgentService def build_session_factory() -> sessionmaker[Session]: @@ -891,6 +898,7 @@ def test_application_preview_action_submits_without_orchestrator_run(monkeypatch "x-auth-name": "Zhang San", "x-auth-employee-no": "E10001", "x-auth-role-codes": "user", + "x-auth-department": "Marketing", }, json={ "source": "user_message", @@ -948,6 +956,83 @@ def test_application_preview_action_submits_without_orchestrator_run(monkeypatch assert claim is not None assert claim.status == "submitted" assert claim.employee_name == "张三" + reservation = db.scalar( + select(BudgetReservation).where( + BudgetReservation.source_type == "application", + BudgetReservation.source_id == claim.id, + BudgetReservation.source_status == "active", + ) + ) + assert reservation is not None + assert reservation.amount == Decimal("1000.00") + case_link = db.scalar( + select(ExpenseCaseLink).where( + ExpenseCaseLink.resource_type == "expense_claim", + ExpenseCaseLink.resource_id == claim.id, + ) + ) + assert case_link is not None + event = db.scalar( + select(BusinessEvent).where( + BusinessEvent.aggregate_id == claim.id, + BusinessEvent.event_type == "application_submitted", + ) + ) + assert event is not None + assert event.expense_case_id == case_link.expense_case_id + assert event.payload_json["previous_status"] == "draft" + assert event.payload_json["next_status"] == "submitted" + assert any( + isinstance(flag, dict) + and flag.get("event_type") == "expense_application_submission" + for flag in list(claim.risk_flags_json or []) + ) + + +def test_application_direct_submit_rolls_back_budget_when_case_event_fails( + monkeypatch, +) -> None: + session_factory = build_session_factory() + + def fail_event(*_args, **_kwargs): + raise RuntimeError("模拟费用事件写入失败") + + monkeypatch.setattr(ExpenseCaseService, "record_claim_event", fail_event) + request = UserAgentRequest( + run_id="application-submit-rollback", + user_id="rollback@example.com", + message="直接提交", + ontology=OntologyParseResult(run_id="application-submit-rollback"), + context_json={ + "session_type": "application", + "name": "回滚测试用户", + "department_name": "Test Department", + }, + ) + facts = { + "application_type": "差旅费用申请", + "time": "2026-07-01 至 2026-07-03", + "location": "北京", + "reason": "验证提交失败回滚", + "amount": "1000元", + } + + with session_factory() as db: + with pytest.raises(RuntimeError, match="模拟费用事件写入失败"): + UserAgentService(db)._create_expense_application_record( + request, + facts, + submit=True, + ) + + # 模拟外层捕获异常后继续记录工具日志并提交，失败草稿仍不能被顺带落库。 + db.commit() + assert list(db.scalars(select(ExpenseClaim)).all()) == [] + assert list(db.scalars(select(BudgetAllocation)).all()) == [] + assert list(db.scalars(select(BudgetReservation)).all()) == [] + assert list(db.scalars(select(BudgetTransaction)).all()) == [] + assert list(db.scalars(select(ExpenseCaseLink)).all()) == [] + assert list(db.scalars(select(BusinessEvent)).all()) == [] def test_application_preview_action_saves_draft_with_detail_reference(monkeypatch, tmp_path) -> None: @@ -1026,3 +1111,12 @@ def test_application_preview_action_saves_draft_with_detail_reference(monkeypatc assert claim.approval_stage == "待提交" assert claim.submitted_at is None assert claim.employee_name == "张三" + assert db.scalar( + select(BudgetReservation).where(BudgetReservation.source_id == claim.id) + ) is None + assert db.scalar( + select(BusinessEvent).where(BusinessEvent.aggregate_id == claim.id) + ) is None + assert db.scalar( + select(ExpenseCaseLink).where(ExpenseCaseLink.resource_id == claim.id) + ) is None diff --git a/server/tests/test_user_agent_service.py b/server/tests/test_user_agent_service.py index 3921b06..5eb1bbf 100644 --- a/server/tests/test_user_agent_service.py +++ b/server/tests/test_user_agent_service.py @@ -627,7 +627,7 @@ def test_user_agent_application_submit_enters_leader_review() -> None: response = build_application_user_agent_response( db, "确认提交", - context_overrides={"manager_name": "陈硕"}, + context_overrides={"manager_name": "陈硕", "department_name": "技术部"}, history=[ {"role": "user", "content": initial_message}, {"role": "user", "content": "飞机"}, @@ -683,7 +683,7 @@ def test_user_agent_application_submit_blocks_duplicate_business_time() -> None: first_response = build_application_user_agent_response( db, "确认提交", - context_overrides={"manager_name": "陈硕"}, + context_overrides={"manager_name": "陈硕", "department_name": "技术部"}, history=history, ) first_claim = application_claim_query(db).one() @@ -691,7 +691,7 @@ def test_user_agent_application_submit_blocks_duplicate_business_time() -> None: second_response = build_application_user_agent_response( db, "确认提交", - context_overrides={"manager_name": "陈硕"}, + context_overrides={"manager_name": "陈硕", "department_name": "技术部"}, history=history, ) @@ -742,6 +742,7 @@ def test_user_agent_application_submit_blocks_overlapping_travel_dates() -> None "确认提交", context_overrides={ "manager_name": "向万红", + "department_name": "技术部", "application_preview": { "fields": { "applicationType": "差旅费用申请", @@ -772,6 +773,7 @@ def test_user_agent_application_submit_normalizes_location_mixed_with_business_c "确认提交", context_overrides={ "manager_name": "向万红", + "department_name": "技术部", "application_preview": { "fields": { "applicationType": "差旅费用申请",