feat(claim): 重构报销审批流并收敛风险标记

- 直属领导兼任部门 P8 预算审批人时合并预算审批，直接流转至财务审批
- 预算超过警戒值时强制要求预算管理者填写审批意见
- 新增风险标记去重工具，消除各审核阶段重复风险卡片
- 新增工作流修复 Mixin，纠正重复预算审批阶段的历史数据
- 收紧单据删除权限至 admin，放宽预算分析可见范围至当前审核人
- 提交校验放宽已上传票据条目的 OCR 字段缺失并忽略尾部占位条目

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

@@ -187,13 +187,13 @@ def get_expense_claim_budget_analysis(
     current_user: CurrentUser,
 ) -> BudgetClaimAnalysisRead:
     service = ExpenseClaimService(db)
-    if not service.can_view_budget_analysis(current_user):
-        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="只有预算监控员或高级财务人员可以查看预算分析。")
     claim = service.get_claim(claim_id, current_user)
     if claim is None:
+        if not service.can_view_budget_analysis(current_user):
+            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="只有当前审核人、该部门预算监控员或高级财务人员可以查看预算分析。")
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Claim not found")
     if not service.can_view_budget_analysis(current_user, claim):
-        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="只有该部门 P8 预算监控员或高级财务人员可以查看预算分析。")
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="只有当前审核人、该部门预算监控员或高级财务人员可以查看预算分析。")
     return BudgetService(db).analyze_claim_budget(claim)
 
 

server/src/app/services/expense_claim_access_policy.py

@@ -28,7 +28,6 @@ APPROVAL_VISIBLE_CLAIM_ROLE_CODES = {"manager", "approver"}
 BUDGET_APPROVAL_ROLE_CODES = {"budget_monitor", "executive"}
 BUDGET_MONITOR_ROLE_CODE = "budget_monitor"
 BUDGET_MONITOR_APPROVAL_GRADE = "P8"
-CLAIM_DELETE_ROLE_CODES = {"executive"}
 ARCHIVED_CLAIM_STATUSES = ("approved", "completed", "paid")
 APPLICATION_ARCHIVED_STAGES = (APPLICATION_ARCHIVE_STAGE,)
 ARCHIVED_REIMBURSEMENT_STAGES = (
@@ -95,9 +94,7 @@ class ExpenseClaimAccessPolicy:
 
     @staticmethod
     def has_claim_delete_access(current_user: CurrentUserContext) -> bool:
-        if current_user.is_admin:
+        return bool(current_user.is_admin)
-            return True
-        return bool(ExpenseClaimAccessPolicy.normalize_role_codes(current_user) & CLAIM_DELETE_ROLE_CODES)
 
     @staticmethod
     def is_archived_claim(claim: ExpenseClaim) -> bool:

server/src/app/services/expense_claim_approval_flow.py

@@ -2,9 +2,11 @@ from __future__ import annotations
 
 import uuid
 from datetime import UTC, datetime
+from decimal import Decimal, InvalidOperation
 from typing import Any
 
 from app.api.deps import CurrentUserContext
+from app.services.budget import BudgetService
 from app.services.expense_claim_workflow_constants import (
     APPLICATION_LINK_STATUS_STAGE,
     BUDGET_MANAGER_APPROVAL_STAGE,
@@ -76,7 +78,16 @@ class ExpenseClaimApprovalFlowMixin:
                     next_stage = APPLICATION_LINK_STATUS_STAGE
                     default_message = "{operator} 已确认直属领导审核，系统判断预算充足且无风险，申请流程完成并生成报销草稿。"
             else:
-                if requires_budget_review:
+                merged_budget_approval = (
+                    requires_budget_review
+                    and self._access_policy.is_department_p8_budget_monitor(current_user, claim)
+                )
+                if merged_budget_approval:
+                    label = "领导及预算审核通过"
+                    next_status = "submitted"
+                    next_stage = FINANCE_APPROVAL_STAGE
+                    default_message = "{operator} 已完成直属领导和预算管理者审核，流转至{next_stage}。"
+                elif requires_budget_review:
                     next_budget_manager = self._access_policy.resolve_department_budget_manager(claim)
                     if next_budget_manager is None:
                         raise ValueError("未找到同部门 P8 预算审批人，无法流转预算审批。请先配置预算审批人。")
@@ -120,6 +131,19 @@ class ExpenseClaimApprovalFlowMixin:
             raise ValueError("当前节点不支持审批通过。")
 
         approval_opinion = str(opinion or "").strip()
+        if (
+            previous_stage == BUDGET_MANAGER_APPROVAL_STAGE
+            and self._budget_approval_opinion_required(claim)
+            and not approval_opinion
+        ):
+            raise ValueError("预算已超过警戒值，预算管理者需填写审批意见后才能通过。")
+        if (
+            previous_stage == DIRECT_MANAGER_APPROVAL_STAGE
+            and merged_budget_approval
+            and self._budget_approval_opinion_required(claim)
+            and not approval_opinion
+        ):
+            raise ValueError("预算已超过警戒值，预算管理者需填写审批意见后才能通过。")
         if previous_stage in {DIRECT_MANAGER_APPROVAL_STAGE, BUDGET_MANAGER_APPROVAL_STAGE} and not approval_opinion:
             approval_opinion = "同意"
 
@@ -327,3 +351,28 @@ class ExpenseClaimApprovalFlowMixin:
             if opinion:
                 return opinion
         return ""
+
+    def _budget_approval_opinion_required(self, claim) -> bool:
+        budget_result = BudgetService(self.db).analyze_claim_budget(claim)
+        metrics = budget_result.get("metrics") if isinstance(budget_result.get("metrics"), dict) else {}
+        context = (
+            budget_result.get("budget_context")
+            if isinstance(budget_result.get("budget_context"), dict)
+            else {}
+        )
+
+        over_budget_amount = self._budget_decimal(metrics.get("over_budget_amount"))
+        if over_budget_amount > Decimal("0.00"):
+            return True
+
+        after_usage_rate = self._budget_decimal(metrics.get("after_usage_rate"))
+        claim_amount_ratio = self._budget_decimal(metrics.get("claim_amount_ratio"))
+        warning_threshold = self._budget_decimal(context.get("warning_threshold") or "80.00")
+        return max(after_usage_rate, claim_amount_ratio) >= warning_threshold
+
+    @staticmethod
+    def _budget_decimal(value: Any) -> Decimal:
+        try:
+            return Decimal(str(value if value is not None else "0")).quantize(Decimal("0.01"))
+        except (InvalidOperation, ValueError):
+            return Decimal("0.00")

server/src/app/services/expense_claim_item_sync.py

@@ -641,6 +641,12 @@ class ExpenseClaimItemSyncMixin:
         issues: list[str] = []
         claim_location_required = self._is_location_required_expense_type(claim.expense_type)
         claim_min_attachment_count = self._resolve_claim_required_attachment_count(claim)
+        substantive_items = [
+            item
+            for item in list(claim.items or [])
+            if str(item.item_type or "").strip().lower() not in SYSTEM_GENERATED_ITEM_TYPES
+            and not self._is_submission_placeholder_item(item)
+        ]
 
         if self._is_missing_value(claim.employee_name):
             issues.append("申请人未完善")
@@ -658,28 +664,39 @@ class ExpenseClaimItemSyncMixin:
             issues.append("发生时间未完善")
         if int(claim.invoice_count or 0) < claim_min_attachment_count:
             issues.append("票据附件数量不足")
-        if not claim.items:
+        if not substantive_items:
             issues.append("费用明细不能为空")
 
         for index, item in enumerate(claim.items, start=1):
             prefix = f"费用明细第 {index} 条"
             is_system_generated = str(item.item_type or "").strip().lower() in SYSTEM_GENERATED_ITEM_TYPES
+            if is_system_generated or self._is_submission_placeholder_item(item):
+                continue
             item_location_required = self._is_location_required_expense_type(item.item_type or claim.expense_type)
-            if item.item_date is None:
+            item_has_attachment = not self._is_missing_value(item.invoice_id)
+            if not item_has_attachment and item.item_date is None:
                 issues.append(f"{prefix}缺少日期")
             if self._is_missing_value(item.item_type):
                 issues.append(f"{prefix}缺少费用项目")
-            if self._is_missing_value(item.item_reason):
+            if not item_has_attachment and self._is_missing_value(item.item_reason):
                 issues.append(f"{prefix}缺少说明")
-            if item_location_required and self._is_missing_value(item.item_location):
+            if not item_has_attachment and item_location_required and self._is_missing_value(item.item_location):
                 issues.append(f"{prefix}缺少地点")
-            if item.item_amount is None or item.item_amount <= Decimal("0.00"):
+            if not item_has_attachment and (item.item_amount is None or item.item_amount <= Decimal("0.00")):
                 issues.append(f"{prefix}缺少金额")
-            if self._is_attachment_required_item_type(item.item_type) and self._is_missing_value(item.invoice_id):
+            if self._is_attachment_required_item_type(item.item_type) and not item_has_attachment:
                 issues.append(f"{prefix}缺少票据标识")
 
         return issues
 
+    def _is_submission_placeholder_item(self, item: ExpenseClaimItem) -> bool:
+        if not self._is_missing_value(item.invoice_id):
+            return False
+        missing_reason = self._is_missing_value(item.item_reason)
+        missing_location = self._is_missing_value(item.item_location)
+        missing_amount = item.item_amount is None or item.item_amount <= Decimal("0.00")
+        return missing_reason and missing_location and missing_amount
+
     def _is_location_required_expense_type(self, expense_type: str | None) -> bool:
         policy = self._get_expense_scene_policy(expense_type)
         if policy is None:

server/src/app/services/expense_claim_pagination.py

@@ -30,6 +30,7 @@ class ExpenseClaimPaginationMixin:
         )
         stmt = self._access_policy.apply_claim_scope(stmt, current_user)
         result = paginate_select(self.db, stmt, page=page, page_size=page_size)
+        self._repair_duplicate_budget_approval_stages(result.items)
         self._access_policy.attach_budget_approval_snapshots(result.items)
         return result
 
@@ -46,6 +47,7 @@ class ExpenseClaimPaginationMixin:
         )
         stmt = self._access_policy.apply_approval_claim_scope(stmt, current_user)
         result = paginate_select(self.db, stmt, page=page, page_size=page_size)
+        self._repair_duplicate_budget_approval_stages(result.items)
         self._access_policy.attach_budget_approval_snapshots(result.items)
         return result
 

server/src/app/services/expense_claim_platform_risk.py

@@ -23,6 +23,7 @@ from app.services.expense_rule_runtime import (
     RuntimeTravelPolicy,
 )
 from app.services.expense_type_keywords import resolve_expense_type_code_from_text
+from app.services.expense_claim_risk_flags import dedupe_claim_risk_flags
 from app.services.expense_claim_platform_route_risk import resolve_multi_city_related_item_ids
 from app.services.expense_claim_platform_risk_flag import build_platform_risk_flag
 from app.services.expense_claim_platform_text_risk import (
@@ -79,6 +80,13 @@ class ExpenseClaimPlatformRiskMixin:
                 continue
 
             flags.append(flag)
+
+        flags = [
+            flag
+            for flag in dedupe_claim_risk_flags(flags)
+            if isinstance(flag, dict)
+        ]
+        for flag in flags:
             severity = str(flag.get("severity") or "").strip().lower()
             action = str(flag.get("action") or "").strip().lower()
             if severity in {"high", "critical"} or action == "block":

server/src/app/services/expense_claim_pre_review.py

@@ -6,6 +6,7 @@ from typing import Any
 from app.api.deps import CurrentUserContext
 from app.models.financial_record import ExpenseClaim
 from app.services.expense_claim_errors import ExpenseClaimSubmissionBlockedError
+from app.services.expense_claim_risk_flags import dedupe_claim_risk_flags
 from app.services.expense_claim_risk_stage import risk_business_stage_for_claim, with_risk_business_stage
 
 
@@ -48,7 +49,9 @@ class ExpenseClaimPreReviewMixin:
                 claim,
                 business_stage="expense_application",
             )
-            review_flags = [*preserved_flags, *list(application_review.get("flags") or [])]
+            review_flags = dedupe_claim_risk_flags(
+                [*preserved_flags, *list(application_review.get("flags") or [])]
+            )
             blocking_count = self._count_ai_pre_review_blocking_risks(review_flags)
             passed = blocking_count <= 0
         else:
@@ -168,7 +171,9 @@ class ExpenseClaimPreReviewMixin:
                 claim,
                 business_stage="expense_application",
             )
-            review_flags = [*preserved_flags, *list(application_review.get("flags") or [])]
+            review_flags = dedupe_claim_risk_flags(
+                [*preserved_flags, *list(application_review.get("flags") or [])]
+            )
         else:
             review_result = self._run_ai_submission_review(claim)
             review_flags = list(review_result.get("risk_flags") or [])

server/src/app/services/expense_claim_risk_flags.py

@@ -0,0 +1,147 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+_SEVERITY_WEIGHT = {
+    "critical": 0,
+    "high": 1,
+    "medium": 2,
+    "low": 3,
+    "info": 4,
+    "pass": 5,
+}
+
+
+def _text(value: Any) -> str:
+    return str(value or "").strip()
+
+
+def _severity_weight(flag: dict[str, Any]) -> int:
+    severity = _text(flag.get("severity") or flag.get("tone") or flag.get("level")).lower()
+    return _SEVERITY_WEIGHT.get(severity, 9)
+
+
+def _list_values(value: Any) -> list[Any]:
+    if isinstance(value, list):
+        return value
+    if _text(value):
+        return [value]
+    return []
+
+
+def _item_ids(flag: dict[str, Any]) -> list[str]:
+    values: list[Any] = [
+        flag.get("item_id"),
+        flag.get("itemId"),
+        *_list_values(flag.get("item_ids")),
+        *_list_values(flag.get("itemIds")),
+    ]
+    item_ids: list[str] = []
+    seen: set[str] = set()
+    for value in values:
+        item_id = _text(value)
+        if not item_id or item_id in seen:
+            continue
+        seen.add(item_id)
+        item_ids.append(item_id)
+    return item_ids
+
+
+def _card_text(flag: dict[str, Any]) -> str:
+    return " ".join(
+        _text(flag.get(key))
+        for key in (
+            "label",
+            "title",
+            "name",
+            "message",
+            "summary",
+            "reason",
+            "description",
+            "rule_code",
+            "ruleCode",
+        )
+    )
+
+
+def _duplicate_group(flag: dict[str, Any]) -> str:
+    text = _card_text(flag)
+    if any(
+        term in text
+        for term in (
+            "多城市行程",
+            "中转",
+            "多地拜访",
+            "改签",
+            "多地出差",
+            "后续行程",
+            "行程终点异常",
+            "连续闭环",
+        )
+    ) and any(
+        term in text for term in ("待说明", "未说明", "缺少说明", "原因", "说明", "不一致", "异常")
+    ):
+        return "route-explanation"
+    if any(term in text for term in ("票据城市", "申报目的地", "行程城市", "酒店票据地点")) and any(
+        term in text for term in ("不一致", "不匹配", "额外中转", "绕行")
+    ):
+        return "travel-city-consistency"
+    if any(term in text for term in ("住宿", "酒店", "宾馆")) and any(
+        term in text for term in ("超标", "超出", "报销标准", "住宿标准", "差标")
+    ):
+        return "hotel-over-standard"
+
+    rule_code = _text(flag.get("rule_code") or flag.get("ruleCode"))
+    if rule_code:
+        return f"rule:{rule_code}"
+    return ""
+
+
+def _same_stage(left: dict[str, Any], right: dict[str, Any]) -> bool:
+    left_stage = _text(left.get("business_stage") or left.get("businessStage"))
+    right_stage = _text(right.get("business_stage") or right.get("businessStage"))
+    return not left_stage or not right_stage or left_stage == right_stage
+
+
+def _same_issue(left: dict[str, Any], right: dict[str, Any]) -> bool:
+    if not _same_stage(left, right):
+        return False
+
+    left_group = _duplicate_group(left)
+    if not left_group or left_group != _duplicate_group(right):
+        return False
+
+    left_items = _item_ids(left)
+    right_items = _item_ids(right)
+    if not left_items or not right_items:
+        return True
+    return any(item_id in right_items for item_id in left_items)
+
+
+def dedupe_claim_risk_flags(flags: list[Any] | None) -> list[Any]:
+    """Remove lower-severity duplicate business risk flags at the data source."""
+
+    normalized_flags = list(flags or [])
+    deduped: list[Any] = []
+    for index, flag in enumerate(normalized_flags):
+        if not isinstance(flag, dict):
+            deduped.append(flag)
+            continue
+
+        current_weight = _severity_weight(flag)
+        is_shadowed = False
+        for other_index, other in enumerate(normalized_flags):
+            if other_index == index or not isinstance(other, dict):
+                continue
+            if not _same_issue(flag, other):
+                continue
+
+            other_weight = _severity_weight(other)
+            if other_weight < current_weight or (other_weight == current_weight and other_index < index):
+                is_shadowed = True
+                break
+
+        if not is_shadowed:
+            deduped.append(flag)
+    return deduped

server/src/app/services/expense_claim_risk_review.py

@@ -15,6 +15,7 @@ from app.services.expense_claim_constants import (
 from app.services.expense_claim_item_sync import ExpenseClaimItemSyncMixin
 from app.services.expense_claim_platform_risk import ExpenseClaimPlatformRiskMixin
 from app.services.expense_claim_policy_review import ExpenseClaimPolicyReviewMixin
+from app.services.expense_claim_risk_flags import dedupe_claim_risk_flags
 from app.services.expense_claim_risk_stage import with_risk_business_stage
 from app.services.risk_observations import RiskObservationService
 
@@ -103,11 +104,12 @@ class ExpenseClaimRiskReviewMixin(
                 )
 
         review_flags = [with_risk_business_stage(flag, "reimbursement") for flag in review_flags]
+        final_risk_flags = dedupe_claim_risk_flags([*preserved_flags, *review_flags])
 
         return {
             "status": "submitted",
             "approval_stage": "直属领导审批",
-            "risk_flags": preserved_flags + review_flags,
+            "risk_flags": final_risk_flags,
             "message": (
                 f"报销单 {claim.claim_no} 已完成自动检测，"
                 f"现已提交给直属领导 {manager_name or '审批人'} 审批。"

server/src/app/services/expense_claim_workflow_repair.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+from app.models.financial_record import ExpenseClaim
+from app.services.expense_claim_risk_stage import risk_business_stage_for_claim, with_risk_business_stage
+from app.services.expense_claim_workflow_constants import (
+    BUDGET_MANAGER_APPROVAL_STAGE,
+    DIRECT_MANAGER_APPROVAL_STAGE,
+    FINANCE_APPROVAL_STAGE,
+)
+
+
+class ExpenseClaimWorkflowRepairMixin:
+    def _repair_duplicate_budget_approval_stages(self, claims: list[ExpenseClaim]) -> None:
+        repaired_claims = [
+            claim
+            for claim in claims
+            if claim is not None and self._repair_duplicate_budget_approval_stage(claim)
+        ]
+        if not repaired_claims:
+            return
+
+        self.db.commit()
+        for claim in repaired_claims:
+            self.db.refresh(claim)
+
+    def _repair_duplicate_budget_approval_stage(self, claim: ExpenseClaim) -> bool:
+        if self._is_expense_application_claim(claim):
+            return False
+        if str(claim.status or "").strip().lower() != "submitted":
+            return False
+        if str(claim.approval_stage or "").strip() != BUDGET_MANAGER_APPROVAL_STAGE:
+            return False
+        if self._has_duplicate_budget_stage_repair_flag(claim):
+            return False
+
+        approval_event = self._find_duplicate_budget_handoff_event(claim)
+        if approval_event is None:
+            return False
+
+        claim.approval_stage = FINANCE_APPROVAL_STAGE
+        claim.risk_flags_json = [
+            *list(claim.risk_flags_json or []),
+            self._build_duplicate_budget_stage_repair_flag(approval_event),
+        ]
+        return True
+
+    def _find_duplicate_budget_handoff_event(self, claim: ExpenseClaim) -> dict[str, Any] | None:
+        flags = [
+            flag
+            for flag in list(claim.risk_flags_json or [])
+            if isinstance(flag, dict)
+            and str(flag.get("source") or "").strip() == "manual_approval"
+            and str(flag.get("event_type") or "").strip() == "expense_claim_approval"
+            and str(flag.get("previous_approval_stage") or "").strip() == DIRECT_MANAGER_APPROVAL_STAGE
+            and str(flag.get("next_approval_stage") or "").strip() == BUDGET_MANAGER_APPROVAL_STAGE
+        ]
+        for flag in reversed(flags):
+            operator = self._normalize_repair_identity(flag.get("operator"))
+            next_approver_name = self._normalize_repair_identity(flag.get("next_approver_name"))
+            if operator and next_approver_name and operator == next_approver_name:
+                return flag
+
+            budget_manager = self._access_policy.resolve_department_budget_manager(claim)
+            if budget_manager is None:
+                continue
+            if operator and operator == self._normalize_repair_identity(budget_manager.name):
+                return flag
+        return None
+
+    def _has_duplicate_budget_stage_repair_flag(self, claim: ExpenseClaim) -> bool:
+        return any(
+            isinstance(flag, dict)
+            and str(flag.get("source") or "").strip() == "approval_flow_repair"
+            and str(flag.get("event_type") or "").strip() == "duplicate_budget_approval_stage_repaired"
+            for flag in list(claim.risk_flags_json or [])
+        )
+
+    def _build_duplicate_budget_stage_repair_flag(self, approval_event: dict[str, Any]) -> dict[str, Any]:
+        return with_risk_business_stage(
+            {
+                "source": "approval_flow_repair",
+                "event_type": "duplicate_budget_approval_stage_repaired",
+                "severity": "info",
+                "label": "重复预算审批已跳过",
+                "message": "系统识别直属领导与预算管理者为同一人，已跳过重复预算审批并流转至财务审批。",
+                "previous_approval_stage": BUDGET_MANAGER_APPROVAL_STAGE,
+                "next_approval_stage": FINANCE_APPROVAL_STAGE,
+                "related_approval_event_id": approval_event.get("approval_event_id"),
+                "budget_approval_merged": True,
+                "budget_approval_merged_reason": "direct_manager_is_department_budget_approver",
+                "created_at": datetime.now(UTC).isoformat(),
+            },
+            risk_business_stage_for_claim(is_application_claim=False),
+        )
+
+    @staticmethod
+    def _normalize_repair_identity(value: Any) -> str:
+        return str(value or "").strip().lower()

server/src/app/services/expense_claims.py

@@ -49,6 +49,7 @@ from app.services.expense_claim_attachment_document import ExpenseClaimAttachmen
 from app.services.expense_claim_attachment_operations import ExpenseClaimAttachmentOperationsMixin
 from app.services.expense_claim_budget_flow import ExpenseClaimBudgetFlowMixin
 from app.services.expense_claim_workflow_constants import DIRECT_MANAGER_APPROVAL_STAGE
+from app.services.expense_claim_workflow_repair import ExpenseClaimWorkflowRepairMixin
 from app.services.expense_claim_document_item_builder import ExpenseClaimDocumentItemBuilderMixin
 from app.services.expense_claim_document_parsing import ExpenseClaimDocumentParsingMixin
 from app.services.expense_claim_draft_flow import ExpenseClaimDraftFlowMixin
@@ -58,6 +59,7 @@ from app.services.expense_claim_pagination import ExpenseClaimPaginationMixin
 from app.services.expense_claim_pre_review import ExpenseClaimPreReviewMixin
 from app.services.expense_claim_ontology_resolvers import ExpenseClaimOntologyResolverMixin
 from app.services.expense_claim_read_model import ExpenseClaimReadModelMixin
+from app.services.expense_claim_risk_flags import dedupe_claim_risk_flags
 from app.services.expense_claim_risk_stage import with_risk_business_stage
 from app.services.expense_claim_review_preview import ExpenseClaimReviewPreviewMixin
 from app.services.receipt_folder import ReceiptFolderService
@@ -156,6 +158,7 @@ class ExpenseClaimService(
     ExpenseClaimAttachmentAnalysisMixin,
     ExpenseClaimReadModelMixin,
     ExpenseClaimRiskReviewMixin,
+    ExpenseClaimWorkflowRepairMixin,
 ):
     def __init__(self, db: Session) -> None:
         self.db = db
@@ -210,7 +213,9 @@ class ExpenseClaimService(
             .order_by(ExpenseClaim.created_at.desc(), ExpenseClaim.occurred_at.desc())
         )
         stmt = self._access_policy.apply_claim_scope(stmt, current_user)
-        return self._access_policy.attach_budget_approval_snapshots(list(self.db.scalars(stmt).all()))
+        claims = list(self.db.scalars(stmt).all())
+        self._repair_duplicate_budget_approval_stages(claims)
+        return self._access_policy.attach_budget_approval_snapshots(claims)
 
     def list_approval_claims(self, current_user: CurrentUserContext) -> list[ExpenseClaim]:
         stmt = (
@@ -224,7 +229,9 @@ class ExpenseClaimService(
             .order_by(ExpenseClaim.submitted_at.desc(), ExpenseClaim.created_at.desc())
         )
         stmt = self._access_policy.apply_approval_claim_scope(stmt, current_user)
-        return self._access_policy.attach_budget_approval_snapshots(list(self.db.scalars(stmt).all()))
+        claims = list(self.db.scalars(stmt).all())
+        self._repair_duplicate_budget_approval_stages(claims)
+        return self._access_policy.attach_budget_approval_snapshots(claims)
 
     def list_archived_claims(self, current_user: CurrentUserContext) -> list[ExpenseClaim]:
         stmt = (
@@ -252,7 +259,10 @@ class ExpenseClaimService(
             .where(ExpenseClaim.id == claim_id)
         )
         stmt = self._access_policy.apply_claim_scope(stmt, current_user, include_approval_scope=True)
-        return self._access_policy.attach_approval_snapshot(self.db.scalar(stmt))
+        claim = self.db.scalar(stmt)
+        if claim is not None:
+            self._repair_duplicate_budget_approval_stages([claim])
+        return self._access_policy.attach_approval_snapshot(claim)
 
     def can_view_budget_analysis(self, current_user: CurrentUserContext, claim: ExpenseClaim | None = None) -> bool:
         if claim is None:
@@ -262,6 +272,13 @@ class ExpenseClaimService(
         role_codes = self._access_policy.normalize_role_codes(current_user)
         if "executive" in role_codes:
             return True
+        if (
+            self._access_policy.has_privileged_claim_access(current_user)
+            and not self._access_policy.is_claim_owned_by_current_user(claim, current_user)
+        ):
+            return True
+        if self._access_policy.can_approve_claim(current_user, claim):
+            return True
         if self._access_policy.is_claim_owned_by_current_user(claim, current_user):
             return False
         return self._access_policy.is_department_p8_budget_monitor(current_user, claim)
@@ -545,7 +562,7 @@ class ExpenseClaimService(
                 and str(flag.get("source") or "").strip() == STANDARD_ADJUSTMENT_RISK_SOURCE
             )
         ]
-        claim.risk_flags_json = [*preserved_flags, *adjustment_flags]
+        claim.risk_flags_json = dedupe_claim_risk_flags([*preserved_flags, *adjustment_flags])
         self._sync_claim_from_items(claim)
 
         self.db.commit()
@@ -805,6 +822,7 @@ class ExpenseClaimService(
             claim.approval_stage = DIRECT_MANAGER_APPROVAL_STAGE
             claim.submitted_at = datetime.now(UTC)
 
+        claim.risk_flags_json = dedupe_claim_risk_flags(claim.risk_flags_json)
 
         self.db.commit()
         self.db.refresh(claim)
@@ -829,9 +847,7 @@ class ExpenseClaimService(
 
     def delete_claim(self, claim_id: str, current_user: CurrentUserContext) -> ExpenseClaim | None:
         claim = self.get_claim(claim_id, current_user)
-        if claim is None and (
+        if claim is None and current_user.is_admin:
-            current_user.is_admin or self._access_policy.has_archive_center_access(current_user)
-        ):
             candidate_claim = self.db.scalar(
                 select(ExpenseClaim)
                 .options(
@@ -841,13 +857,14 @@ class ExpenseClaimService(
                 )
                 .where(ExpenseClaim.id == claim_id)
             )
-            if candidate_claim is not None and (
+            if candidate_claim is not None:
-                current_user.is_admin or self._access_policy.is_archived_claim(candidate_claim)
-            ):
                 claim = candidate_claim
         if claim is None:
             return None
 
+        if not self._access_policy.has_claim_delete_access(current_user):
+            raise ValueError("只有 admin 管理员可以删除单据。")
+
         if self._access_policy.is_archived_claim(claim) and not current_user.is_admin:
             raise ValueError("已归档单据不能删除，只有高级管理员可以执行删除。")