fix(expenses): unify AI application submission transaction

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 关联不会残留部分成功数据。

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` 通过；未修改数据库结构，未对持久化开发数据库执行迁移。

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 摘要、伪造身份头、过期/撤销、登出原子收尾、业务经理越权和初始化后匿名重配置拒绝。

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

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,

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

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": "差旅费用申请",