X-Financial/server/tests/test_attachment_association_jobs.py

from __future__ import annotations

import base64
from collections.abc import Generator
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal
from threading import Lock, Thread
from time import sleep

from auth_helpers import install_legacy_header_auth_override
from fastapi.testclient import TestClient
from sqlalchemy import select
from sqlalchemy.orm import Session, selectinload

from app.api.deps import CurrentUserContext, get_db
from app.api.v1.endpoints import attachment_association_jobs as attachment_jobs_endpoint
from app.core.config import get_settings
from app.main import create_app
from app.models.attachment_association_job import AttachmentAssociationJob
from app.models.employee import Employee
from app.models.expense_case import BusinessEvent, ExpenseCase, ExpenseCaseLink
from app.models.financial_record import ExpenseClaim, ExpenseClaimItem
from app.schemas.attachment_association_job import AttachmentAssociationJobCreate
from app.schemas.ocr import OcrRecognizeBatchRead, OcrRecognizeDocumentRead, OcrRecognizeFieldRead
from app.services.attachment_association_job_store import (
    claim_persistent_job,
    update_persistent_job,
)
from app.services.attachment_association_jobs import (
    clear_attachment_association_jobs_for_tests,
    create_attachment_association_job,
    get_attachment_association_job,
    run_attachment_association_job,
)
from app.services.expense_cases import ExpenseCaseService
from app.services.expense_claim_attachment_storage import ExpenseClaimAttachmentStorage
from app.services.expense_claims import ExpenseClaimService
from app.services.expense_receipt_association import ExpenseReceiptAssociationService
from app.services.ocr import OcrService
from app.services.receipt_folder import ReceiptFolderService
from app.test_helpers.db import build_in_memory_session_factory


def build_client(monkeypatch) -> tuple[TestClient, object]:
    session_factory = build_in_memory_session_factory()
    app = create_app()
    install_legacy_header_auth_override(app)

    def override_db() -> Generator[Session, None, None]:
        db = session_factory()
        try:
            yield db
        finally:
            db.close()

    app.dependency_overrides[get_db] = override_db
    monkeypatch.setattr(attachment_jobs_endpoint, "get_session_factory", lambda: session_factory)
    return TestClient(app), session_factory


def seed_travel_claim(db: Session) -> ExpenseClaim:
    employee = Employee(
        id="emp-bg-association",
        employee_no="E10001",
        name="张三",
        email="zhangsan@example.com",
        position="实施顾问",
        grade="P4",
    )
    claim = ExpenseClaim(
        id="claim-bg-association",
        claim_no="BX-20260220-001",
        employee_id=employee.id,
        employee_name=employee.name,
        department_id="dept-delivery",
        department_name="交付部",
        project_code=None,
        expense_type="travel",
        reason="辅助国网仿生产服务器部署，武汉往返上海",
        location="上海",
        amount=Decimal("0.00"),
        currency="CNY",
        invoice_count=0,
        occurred_at=datetime(2026, 2, 20, tzinfo=UTC),
        submitted_at=None,
        status="draft",
        approval_stage="待提交",
        risk_flags_json=[],
    )
    item = ExpenseClaimItem(
        id="item-bg-association-1",
        claim_id=claim.id,
        item_date=date(2026, 2, 20),
        item_type="train_ticket",
        item_reason="武汉至上海高铁",
        item_location="上海",
        item_amount=Decimal("0.00"),
        invoice_id=None,
    )
    claim.items = [item]
    db.add_all([employee, claim])
    db.commit()
    return claim


def seed_approved_application_for_draft(db: Session, draft: ExpenseClaim) -> ExpenseClaim:
    application = ExpenseClaim(
        id="application-bg-association",
        claim_no="AP-20260220-001",
        employee_id=draft.employee_id,
        employee_name=draft.employee_name,
        department_id=draft.department_id,
        department_name=draft.department_name,
        project_code=draft.project_code,
        expense_type="travel_application",
        reason=draft.reason,
        location=draft.location,
        amount=Decimal("3000.00"),
        currency="CNY",
        invoice_count=0,
        occurred_at=draft.occurred_at,
        submitted_at=draft.occurred_at,
        status="approved",
        approval_stage="已完成",
        risk_flags_json=[],
    )
    db.add(application)
    db.flush()
    case_service = ExpenseCaseService(db)
    expense_case = case_service.ensure_case_for_claim(
        application,
        tenant_id="default",
        relation_type="application",
    )
    case_service.link_claim(
        expense_case,
        draft,
        tenant_id="default",
        relation_type="generated_reimbursement",
    )
    db.commit()
    return application


def seed_standalone_approved_application(db: Session) -> ExpenseClaim:
    employee = Employee(
        id="emp-application-only",
        employee_no="E10001",
        name="张三",
        email="zhangsan@example.com",
        position="实施顾问",
        grade="P4",
    )
    application = ExpenseClaim(
        id="application-only-association",
        claim_no="AP-20260220-ONLY",
        employee_id=employee.id,
        employee_name=employee.name,
        department_id="dept-delivery",
        department_name="交付部",
        project_code=None,
        expense_type="travel_application",
        reason="辅助国网仿生产服务器部署，武汉往返上海",
        location="上海",
        amount=Decimal("3000.00"),
        currency="CNY",
        invoice_count=0,
        occurred_at=datetime(2026, 2, 20, tzinfo=UTC),
        submitted_at=datetime(2026, 2, 18, tzinfo=UTC),
        status="approved",
        approval_stage="已完成",
        risk_flags_json=[],
    )
    db.add_all([employee, application])
    db.flush()
    ExpenseCaseService(db).ensure_case_for_claim(
        application,
        tenant_id="default",
        relation_type="application",
    )
    db.commit()
    return application


def seed_approved_application_with_flag_only(
    db: Session,
    draft: ExpenseClaim,
) -> tuple[ExpenseClaim, ExpenseCase]:
    application = ExpenseClaim(
        id="application-flag-only-association",
        claim_no="AP-20260220-FLAG",
        employee_id=draft.employee_id,
        employee_name=draft.employee_name,
        department_id=draft.department_id,
        department_name=draft.department_name,
        project_code=draft.project_code,
        expense_type="travel_application",
        reason=draft.reason,
        location=draft.location,
        amount=Decimal("3000.00"),
        currency="CNY",
        invoice_count=0,
        occurred_at=draft.occurred_at,
        submitted_at=draft.occurred_at,
        status="approved",
        approval_stage="已完成",
        risk_flags_json=[],
    )
    draft.risk_flags_json = [
        {
            "source": "application_link",
            "application_claim_id": application.id,
            "application_claim_no": application.claim_no,
        }
    ]
    db.add(application)
    db.flush()
    expense_case = ExpenseCaseService(db).ensure_case_for_claim(
        application,
        tenant_id="default",
        relation_type="application",
    )
    db.commit()
    return application, expense_case


def seed_second_matching_draft(db: Session) -> ExpenseClaim:
    claim = ExpenseClaim(
        id="claim-bg-association-2",
        claim_no="BX-20260220-002",
        employee_id="emp-bg-association",
        employee_name="张三",
        department_id="dept-delivery",
        department_name="交付部",
        project_code=None,
        expense_type="travel",
        reason="辅助国网仿生产服务器部署，武汉往返上海",
        location="上海",
        amount=Decimal("0.00"),
        currency="CNY",
        invoice_count=0,
        occurred_at=datetime(2026, 2, 20, tzinfo=UTC),
        submitted_at=None,
        status="draft",
        approval_stage="待提交",
        risk_flags_json=[],
        items=[
            ExpenseClaimItem(
                id="item-bg-association-2",
                item_date=date(2026, 2, 20),
                item_type="train_ticket",
                item_reason="武汉至上海高铁",
                item_location="上海",
                item_amount=Decimal("0.00"),
                invoice_id=None,
            )
        ],
    )
    db.add(claim)
    db.commit()
    return claim


def save_train_receipt(
    *,
    service: ReceiptFolderService,
    current_user: CurrentUserContext,
    filename: str,
    route: str,
    trip_date: str,
) -> str:
    receipt = service.save_receipt(
        filename=filename,
        content=f"fake-pdf-{filename}".encode(),
        media_type="application/pdf",
        current_user=current_user,
        document=OcrRecognizeDocumentRead(
            filename=filename,
            media_type="application/pdf",
            text=f"电子发票（铁路电子客票） {route} {trip_date} 票价 354 元",
            summary=f"铁路电子客票，{route}，票价 354 元。",
            avg_score=0.96,
            line_count=1,
            page_count=1,
            document_type="train_ticket",
            document_type_label="火车/高铁票",
            scene_code="travel",
            scene_label="差旅票据",
            document_fields=[
                OcrRecognizeFieldRead(key="date", label="列车出发时间", value=trip_date),
                OcrRecognizeFieldRead(key="route", label="行程", value=route),
                OcrRecognizeFieldRead(key="amount", label="金额", value="354元"),
            ],
        ),
    )
    return receipt.id


def fake_ocr_recognize(
    self,
    files: list[tuple[str, bytes, str | None]],
) -> OcrRecognizeBatchRead:
    filename = files[0][0]
    return OcrRecognizeBatchRead(
        total_file_count=1,
        success_count=1,
        documents=[
            OcrRecognizeDocumentRead(
                filename=filename,
                media_type=files[0][2] or "application/pdf",
                text="电子发票（铁路电子客票） 武汉 上海 2026-02-20 票价 354 元",
                summary="铁路电子客票，武汉至上海，票价 354 元。",
                avg_score=0.96,
                line_count=1,
                page_count=1,
                document_type="train_ticket",
                document_type_label="火车/高铁票",
                scene_code="travel",
                scene_label="差旅票据",
                document_fields=[
                    OcrRecognizeFieldRead(key="date", label="列车出发时间", value="2026-02-20"),
                    OcrRecognizeFieldRead(key="route", label="行程", value="武汉-上海"),
                    OcrRecognizeFieldRead(key="amount", label="金额", value="354元"),
                ],
            )
        ],
    )


def fake_ocr_recognize_without_preview(
    self,
    files: list[tuple[str, bytes, str | None]],
) -> OcrRecognizeBatchRead:
    return fake_ocr_recognize(self, files)


def test_attachment_association_job_links_receipts_after_conversation_exit(
    monkeypatch, tmp_path
) -> None:
    monkeypatch.setenv("STORAGE_ROOT_DIR", str(tmp_path / "storage"))
    get_settings.cache_clear()
    clear_attachment_association_jobs_for_tests()
    monkeypatch.setattr(OcrService, "recognize_files", fake_ocr_recognize)
    monkeypatch.setattr(
        ExpenseClaimAttachmentStorage, "root", lambda self: tmp_path / "attachments"
    )
    try:
        client, session_factory = build_client(monkeypatch)
        current_user = CurrentUserContext(
            tenant_id="default",
            username="zhangsan@example.com",
            name="张三",
            role_codes=["user"],
            is_admin=False,
            employee_no="E10001",
        )
        with session_factory() as db:
            draft = seed_travel_claim(db)
            seed_approved_application_for_draft(db, draft)

        receipt_service = ReceiptFolderService()
        receipt_ids = [
            save_train_receipt(
                service=receipt_service,
                current_user=current_user,
                filename="2月20 武汉-上海.pdf",
                route="武汉-上海",
                trip_date="2026-02-20",
            ),
            save_train_receipt(
                service=receipt_service,
                current_user=current_user,
                filename="2月23 上海-武汉.pdf",
                route="上海-武汉",
                trip_date="2026-02-23",
            ),
        ]

        headers = {
            "x-auth-username": "zhangsan@example.com",
            "x-auth-name": "Zhang San",
            "x-auth-employee-no": "E10001",
            "x-auth-role-codes": "user",
        }
        response = client.post(
            "/api/v1/reimbursements/attachment-association-jobs",
            headers=headers,
            json={
                "receipt_ids": receipt_ids,
                "prompt": "请帮我处理已上传的附件。",
                "conversation_id": "inline-test",
            },
        )

        assert response.status_code == 202
        job_id = response.json()["job_id"]

        status_response = client.get(
            f"/api/v1/reimbursements/attachment-association-jobs/{job_id}",
            headers=headers,
        )
        assert status_response.status_code == 200
        payload = status_response.json()
        assert payload["status"] == "succeeded"
        assert payload["claim_id"] == "claim-bg-association"
        assert payload["claim_no"] == "BX-20260220-001"
        assert payload["uploaded_count"] == 2
        assert payload["resolution"] == "auto_associated"
        assert payload["requires_confirmation"] is False
        assert payload["expense_case_id"]
        assert payload["application_claim_id"] == "application-bg-association"
        assert payload["application_claim_no"] == "AP-20260220-001"
        assert payload["confidence"] == "high"
        assert payload["match_reasons"]
        assert payload["draft_payload"]["claim_id"] == "claim-bg-association"

        with session_factory() as db:
            claim = db.scalar(
                select(ExpenseClaim)
                .options(selectinload(ExpenseClaim.items))
                .where(ExpenseClaim.id == "claim-bg-association")
            )
            assert claim is not None
            attached_items = [item for item in claim.items if item.invoice_id]
            assert len(attached_items) == 2
            pre_review_flags = [
                flag
                for flag in list(claim.risk_flags_json or [])
                if isinstance(flag, dict) and flag.get("source") == "ai_pre_review"
            ]
            assert len(pre_review_flags) == 1
            assert pre_review_flags[0]["created_at"]
            receipt_links = list(
                db.scalars(
                    select(ExpenseCaseLink).where(ExpenseCaseLink.resource_type == "receipt")
                ).all()
            )
            assert {link.resource_id for link in receipt_links} == set(receipt_ids)
            receipt_events = list(
                db.scalars(
                    select(BusinessEvent).where(BusinessEvent.aggregate_type == "receipt")
                ).all()
            )
            assert len(receipt_events) == 4
            assert {event.event_type for event in receipt_events} == {
                "receipt_received",
                "attachment_associated",
            }
            expense_case = db.scalar(
                select(ExpenseCase).where(ExpenseCase.id == receipt_links[0].expense_case_id)
            )
            assert expense_case is not None
            assert expense_case.current_stage == "claiming"

        linked_receipts = receipt_service.list_receipts(
            current_user=current_user, status_filter="linked"
        )
        assert {item.id for item in linked_receipts} == set(receipt_ids)
        assert {item.linked_claim_id for item in linked_receipts} == {"claim-bg-association"}
    finally:
        clear_attachment_association_jobs_for_tests()
        get_settings.cache_clear()


def test_attachment_association_keeps_receipt_folder_preview_and_fields_after_cache_clear(
    monkeypatch,
    tmp_path,
) -> None:
    preview_bytes = b"receipt-folder-preview-png"
    preview_data_url = f"data:image/png;base64,{base64.b64encode(preview_bytes).decode('ascii')}"

    monkeypatch.setenv("STORAGE_ROOT_DIR", str(tmp_path / "storage"))
    get_settings.cache_clear()
    clear_attachment_association_jobs_for_tests()
    monkeypatch.setattr(OcrService, "recognize_files", fake_ocr_recognize_without_preview)
    monkeypatch.setattr(
        ExpenseClaimAttachmentStorage, "root", lambda self: tmp_path / "attachments"
    )
    try:
        client, session_factory = build_client(monkeypatch)
        current_user = CurrentUserContext(
            tenant_id="default",
            username="zhangsan@example.com",
            name="张三",
            role_codes=["user"],
            is_admin=False,
            employee_no="E10001",
        )
        with session_factory() as db:
            seed_travel_claim(db)

        receipt = ReceiptFolderService().save_receipt(
            filename="2月20 武汉-上海.pdf",
            content=b"%PDF-1.7 fake-ticket",
            media_type="application/pdf",
            current_user=current_user,
            document=OcrRecognizeDocumentRead(
                filename="2月20 武汉-上海.pdf",
                media_type="application/pdf",
                text=(
                    "电子发票（铁路电子客票） 武汉站 G458 上海虹桥站 "
                    "2026年02月20日 07:55开 二等座 票价 354.00"
                ),
                summary="铁路电子客票，武汉-上海，票价 354 元。",
                avg_score=0.96,
                line_count=1,
                page_count=1,
                document_type="train_ticket",
                document_type_label="火车/高铁票",
                scene_code="travel",
                scene_label="差旅票据",
                preview_kind="image",
                preview_data_url=preview_data_url,
                document_fields=[
                    OcrRecognizeFieldRead(
                        key="date", label="列车出发时间", value="2026-02-20 07:55"
                    ),
                    OcrRecognizeFieldRead(key="route", label="行程", value="武汉-上海"),
                    OcrRecognizeFieldRead(key="amount", label="金额", value="354元"),
                ],
            ),
        )
        OcrService.clear_result_cache()

        headers = {
            "x-auth-username": "zhangsan@example.com",
            "x-auth-name": "Zhang San",
            "x-auth-employee-no": "E10001",
            "x-auth-role-codes": "user",
        }
        response = client.post(
            "/api/v1/reimbursements/attachment-association-jobs",
            headers=headers,
            json={
                "receipt_ids": [receipt.id],
                "prompt": "请帮我处理已上传的附件。",
                "conversation_id": "inline-test",
            },
        )
        assert response.status_code == 202
        job_id = response.json()["job_id"]

        status_response = client.get(
            f"/api/v1/reimbursements/attachment-association-jobs/{job_id}",
            headers=headers,
        )
        assert status_response.status_code == 200
        assert status_response.json()["status"] == "succeeded"

        with session_factory() as db:
            claim = db.scalar(
                select(ExpenseClaim)
                .options(selectinload(ExpenseClaim.items))
                .where(ExpenseClaim.id == "claim-bg-association")
            )
            assert claim is not None
            attached_item = next(item for item in claim.items if item.invoice_id)
            metadata = ExpenseClaimService(db).get_claim_item_attachment_meta(
                claim_id=claim.id,
                item_id=attached_item.id,
                current_user=current_user,
            )
            assert metadata is not None
            assert metadata["preview_kind"] == "image"
            assert metadata["document_info"]["document_type"] == "train_ticket"
            assert metadata["document_info"]["document_type_label"] == "火车/高铁票"
            assert {
                (field["label"], field["value"]) for field in metadata["document_info"]["fields"]
            } >= {
                ("列车出发时间", "2026-02-20 07:55"),
                ("行程", "武汉-上海"),
                ("金额", "354元"),
            }

            preview_path, media_type, filename = ExpenseClaimService(
                db
            ).get_claim_item_attachment_preview_content(
                claim_id=claim.id,
                item_id=attached_item.id,
                current_user=current_user,
            )
            assert media_type == "image/png"
            assert filename.endswith(".png")
            assert preview_path.read_bytes() == preview_bytes
    finally:
        clear_attachment_association_jobs_for_tests()
        get_settings.cache_clear()


def test_attachment_meta_repairs_existing_pdf_fallback_from_source_receipt(
    monkeypatch,
    tmp_path,
) -> None:
    preview_bytes = b"legacy-repaired-preview-png"
    preview_data_url = f"data:image/png;base64,{base64.b64encode(preview_bytes).decode('ascii')}"
    monkeypatch.setenv("STORAGE_ROOT_DIR", str(tmp_path / "storage"))
    get_settings.cache_clear()
    monkeypatch.setattr(
        ExpenseClaimAttachmentStorage, "root", lambda self: tmp_path / "attachments"
    )
    try:
        current_user = CurrentUserContext(
            tenant_id="default",
            username="zhangsan@example.com",
            name="张三",
            role_codes=["user"],
            is_admin=False,
            employee_no="E10001",
        )
        client, session_factory = build_client(monkeypatch)
        client.close()

        with session_factory() as db:
            claim = seed_travel_claim(db)
            item = claim.items[0]
            receipt = ReceiptFolderService().save_receipt(
                filename="2月20 武汉-上海.pdf",
                content=b"%PDF-1.7 fake-ticket",
                media_type="application/pdf",
                current_user=current_user,
                document=OcrRecognizeDocumentRead(
                    filename="2月20 武汉-上海.pdf",
                    media_type="application/pdf",
                    text=(
                        "电子发票（铁路电子客票） 武汉站 G458 上海虹桥站 "
                        "2026年02月20日 07:55开 二等座 票价 354.00"
                    ),
                    summary="铁路电子客票，武汉-上海，票价 354 元。",
                    avg_score=0.96,
                    line_count=1,
                    page_count=1,
                    document_type="train_ticket",
                    document_type_label="火车/高铁票",
                    scene_code="travel",
                    scene_label="差旅票据",
                    preview_kind="image",
                    preview_data_url=preview_data_url,
                    document_fields=[
                        OcrRecognizeFieldRead(
                            key="date", label="列车出发时间", value="2026-02-20 07:55"
                        ),
                        OcrRecognizeFieldRead(key="route", label="行程", value="武汉-上海"),
                        OcrRecognizeFieldRead(key="amount", label="金额", value="354元"),
                    ],
                ),
            )

            attachment_dir = tmp_path / "attachments" / claim.id / item.id
            attachment_dir.mkdir(parents=True)
            file_path = attachment_dir / "2月20_武汉-上海.pdf"
            file_path.write_bytes(b"%PDF-1.7 persisted-but-bad-meta")
            storage = ExpenseClaimAttachmentStorage()
            item.invoice_id = storage.to_storage_key(file_path)
            storage.write_meta(
                file_path,
                {
                    "file_name": file_path.name,
                    "storage_key": storage.to_storage_key(file_path),
                    "media_type": "application/pdf",
                    "size_bytes": file_path.stat().st_size,
                    "previewable": True,
                    "preview_kind": "pdf",
                    "preview_storage_key": storage.to_storage_key(file_path),
                    "preview_media_type": "application/pdf",
                    "preview_file_name": file_path.name,
                    "document_info": {
                        "document_type": "other",
                        "document_type_label": "其他单据",
                        "scene_code": "other",
                        "scene_label": "其他票据",
                        "fields": [],
                    },
                    "source_receipt_id": receipt.id,
                },
            )
            db.commit()

            service = ExpenseClaimService(db)
            metadata = service.get_claim_item_attachment_meta(
                claim_id=claim.id,
                item_id=item.id,
                current_user=current_user,
            )
            assert metadata is not None
            assert metadata["preview_kind"] == "image"
            assert metadata["document_info"]["document_type"] == "train_ticket"
            assert metadata["document_info"]["document_type_label"] == "火车/高铁票"
            assert {
                (field["label"], field["value"]) for field in metadata["document_info"]["fields"]
            } >= {
                ("列车出发时间", "2026-02-20 07:55"),
                ("行程", "武汉-上海"),
                ("金额", "354元"),
            }

            preview_path, media_type, filename = service.get_claim_item_attachment_preview_content(
                claim_id=claim.id,
                item_id=item.id,
                current_user=current_user,
            )
            assert media_type == "image/png"
            assert filename.endswith(".png")
            assert preview_path.read_bytes() == preview_bytes
    finally:
        get_settings.cache_clear()


def test_attachment_association_job_requests_confirmation_without_editable_claim(
    monkeypatch,
    tmp_path,
) -> None:
    monkeypatch.setenv("STORAGE_ROOT_DIR", str(tmp_path / "storage"))
    get_settings.cache_clear()
    clear_attachment_association_jobs_for_tests()
    try:
        client, _session_factory = build_client(monkeypatch)
        current_user = CurrentUserContext(
            tenant_id="default",
            username="zhangsan@example.com",
            name="张三",
            role_codes=["user"],
            is_admin=False,
            employee_no="E10001",
        )
        receipt_id = save_train_receipt(
            service=ReceiptFolderService(),
            current_user=current_user,
            filename="2月20 武汉-上海.pdf",
            route="武汉-上海",
            trip_date="2026-02-20",
        )

        headers = {
            "x-auth-username": "zhangsan@example.com",
            "x-auth-name": "Zhang San",
            "x-auth-employee-no": "E10001",
            "x-auth-role-codes": "user",
        }
        response = client.post(
            "/api/v1/reimbursements/attachment-association-jobs",
            headers=headers,
            json={"receipt_ids": [receipt_id], "conversation_id": "inline-empty"},
        )
        assert response.status_code == 202

        status_response = client.get(
            f"/api/v1/reimbursements/attachment-association-jobs/{response.json()['job_id']}",
            headers=headers,
        )
        assert status_response.status_code == 200
        payload = status_response.json()
        assert payload["status"] == "succeeded"
        assert payload["resolution"] == "requires_confirmation"
        assert payload["requires_confirmation"] is True
        assert payload["uploaded_count"] == 0
        assert payload["error"] == ""
        assert "没有找到" in payload["exceptions"][0]
    finally:
        clear_attachment_association_jobs_for_tests()
        get_settings.cache_clear()


def test_confirmation_job_is_re_evaluated_after_draft_is_created(
    monkeypatch,
    tmp_path,
) -> None:
    monkeypatch.setenv("STORAGE_ROOT_DIR", str(tmp_path / "storage"))
    get_settings.cache_clear()
    clear_attachment_association_jobs_for_tests()
    monkeypatch.setattr(OcrService, "recognize_files", fake_ocr_recognize)
    monkeypatch.setattr(
        ExpenseClaimAttachmentStorage, "root", lambda self: tmp_path / "attachments"
    )
    try:
        client, session_factory = build_client(monkeypatch)
        current_user = CurrentUserContext(
            tenant_id="default",
            username="zhangsan@example.com",
            name="张三",
            role_codes=["user"],
            is_admin=False,
            employee_no="E10001",
        )
        receipt_id = save_train_receipt(
            service=ReceiptFolderService(),
            current_user=current_user,
            filename="2月20 武汉-上海.pdf",
            route="武汉-上海",
            trip_date="2026-02-20",
        )
        headers = {
            "x-auth-username": "zhangsan@example.com",
            "x-auth-name": "Zhang San",
            "x-auth-employee-no": "E10001",
            "x-auth-role-codes": "user",
        }

        first_job = client.post(
            "/api/v1/reimbursements/attachment-association-jobs",
            headers=headers,
            json={"receipt_ids": [receipt_id]},
        ).json()
        first_result = client.get(
            f"/api/v1/reimbursements/attachment-association-jobs/{first_job['job_id']}",
            headers=headers,
        ).json()
        assert first_result["resolution"] == "requires_confirmation"

        with session_factory() as db:
            seed_travel_claim(db)

        second_job = client.post(
            "/api/v1/reimbursements/attachment-association-jobs",
            headers=headers,
            json={"receipt_ids": [receipt_id]},
        ).json()
        second_result = client.get(
            f"/api/v1/reimbursements/attachment-association-jobs/{second_job['job_id']}",
            headers=headers,
        ).json()
        third_job = client.post(
            "/api/v1/reimbursements/attachment-association-jobs",
            headers=headers,
            json={"receipt_ids": [receipt_id]},
        ).json()

        assert second_job["job_id"] != first_job["job_id"]
        assert third_job["job_id"] == second_job["job_id"]
        assert second_result["status"] == "succeeded"
        assert second_result["resolution"] == "auto_associated"
        assert second_result["uploaded_count"] == 1
        with session_factory() as db:
            jobs = list(
                db.scalars(
                    select(AttachmentAssociationJob).order_by(AttachmentAssociationJob.generation)
                ).all()
            )
            assert [(job.generation, job.resolution) for job in jobs] == [
                (1, "requires_confirmation"),
                (2, "auto_associated"),
            ]
    finally:
        clear_attachment_association_jobs_for_tests()
        get_settings.cache_clear()


def test_attachment_association_job_returns_application_candidate_without_creating_draft(
    monkeypatch,
    tmp_path,
) -> None:
    monkeypatch.setenv("STORAGE_ROOT_DIR", str(tmp_path / "storage"))
    get_settings.cache_clear()
    clear_attachment_association_jobs_for_tests()
    try:
        client, session_factory = build_client(monkeypatch)
        current_user = CurrentUserContext(
            tenant_id="default",
            username="zhangsan@example.com",
            name="张三",
            role_codes=["user"],
            is_admin=False,
            employee_no="E10001",
        )
        with session_factory() as db:
            application = seed_standalone_approved_application(db)
            application_id = application.id
        receipt_id = save_train_receipt(
            service=ReceiptFolderService(),
            current_user=current_user,
            filename="2月20 武汉-上海.pdf",
            route="武汉-上海",
            trip_date="2026-02-20",
        )
        headers = {
            "x-auth-username": "zhangsan@example.com",
            "x-auth-name": "Zhang San",
            "x-auth-employee-no": "E10001",
            "x-auth-role-codes": "user",
        }
        created = client.post(
            "/api/v1/reimbursements/attachment-association-jobs",
            headers=headers,
            json={"receipt_ids": [receipt_id]},
        ).json()
        result = client.get(
            f"/api/v1/reimbursements/attachment-association-jobs/{created['job_id']}",
            headers=headers,
        ).json()

        assert result["status"] == "succeeded"
        assert result["resolution"] == "requires_confirmation"
        assert result["requires_confirmation"] is True
        assert result["claim_id"] == ""
        assert result["uploaded_count"] == 0
        assert result["candidates"][0]["target_type"] == "approved_application"
        assert result["candidates"][0]["application_claim_id"] == application_id
        with session_factory() as db:
            claims = list(db.scalars(select(ExpenseClaim)).all())
            assert [claim.id for claim in claims] == [application_id]
    finally:
        clear_attachment_association_jobs_for_tests()
        get_settings.cache_clear()


def test_attachment_association_job_is_idempotent_for_same_receipt(monkeypatch, tmp_path) -> None:
    monkeypatch.setenv("STORAGE_ROOT_DIR", str(tmp_path / "storage"))
    get_settings.cache_clear()
    clear_attachment_association_jobs_for_tests()
    monkeypatch.setattr(OcrService, "recognize_files", fake_ocr_recognize)
    monkeypatch.setattr(
        ExpenseClaimAttachmentStorage, "root", lambda self: tmp_path / "attachments"
    )
    try:
        client, session_factory = build_client(monkeypatch)
        current_user = CurrentUserContext(
            tenant_id="default",
            username="zhangsan@example.com",
            name="张三",
            role_codes=["user"],
            is_admin=False,
            employee_no="E10001",
        )
        with session_factory() as db:
            seed_travel_claim(db)
        receipt_id = save_train_receipt(
            service=ReceiptFolderService(),
            current_user=current_user,
            filename="2月20 武汉-上海.pdf",
            route="武汉-上海",
            trip_date="2026-02-20",
        )
        headers = {
            "x-auth-username": "zhangsan@example.com",
            "x-auth-name": "Zhang San",
            "x-auth-employee-no": "E10001",
            "x-auth-role-codes": "user",
        }

        first_job = client.post(
            "/api/v1/reimbursements/attachment-association-jobs",
            headers=headers,
            json={"receipt_ids": [receipt_id]},
        ).json()
        first_result = client.get(
            f"/api/v1/reimbursements/attachment-association-jobs/{first_job['job_id']}",
            headers=headers,
        ).json()
        second_job = client.post(
            "/api/v1/reimbursements/attachment-association-jobs",
            headers=headers,
            json={"receipt_ids": [receipt_id]},
        ).json()
        second_result = client.get(
            f"/api/v1/reimbursements/attachment-association-jobs/{second_job['job_id']}",
            headers=headers,
        ).json()

        assert second_job["job_id"] == first_job["job_id"]
        assert first_result["uploaded_count"] == 1
        assert second_result["error"] == "", second_result["error"]
        assert second_result["resolution"] == "auto_associated", second_result
        assert second_result["uploaded_count"] == 1
        assert second_result["skipped_count"] == 0
        with session_factory() as db:
            claim = db.scalar(
                select(ExpenseClaim)
                .options(selectinload(ExpenseClaim.items))
                .where(ExpenseClaim.id == "claim-bg-association")
            )
            assert claim is not None
            assert len([item for item in claim.items if item.invoice_id]) == 1
            events = list(
                db.scalars(
                    select(BusinessEvent).where(BusinessEvent.aggregate_type == "receipt")
                ).all()
            )
            assert len(events) == 2
    finally:
        clear_attachment_association_jobs_for_tests()
        get_settings.cache_clear()


def test_attachment_association_ambiguous_match_has_no_business_writes(
    monkeypatch, tmp_path
) -> None:
    monkeypatch.setenv("STORAGE_ROOT_DIR", str(tmp_path / "storage"))
    get_settings.cache_clear()
    clear_attachment_association_jobs_for_tests()
    try:
        client, session_factory = build_client(monkeypatch)
        current_user = CurrentUserContext(
            tenant_id="default",
            username="zhangsan@example.com",
            name="张三",
            role_codes=["user"],
            is_admin=False,
            employee_no="E10001",
        )
        with session_factory() as db:
            seed_travel_claim(db)
            seed_second_matching_draft(db)
        receipt_id = save_train_receipt(
            service=ReceiptFolderService(),
            current_user=current_user,
            filename="2月20 武汉-上海.pdf",
            route="武汉-上海",
            trip_date="2026-02-20",
        )
        headers = {
            "x-auth-username": "zhangsan@example.com",
            "x-auth-name": "Zhang San",
            "x-auth-employee-no": "E10001",
            "x-auth-role-codes": "user",
        }
        created = client.post(
            "/api/v1/reimbursements/attachment-association-jobs",
            headers=headers,
            json={"receipt_ids": [receipt_id]},
        ).json()
        result = client.get(
            f"/api/v1/reimbursements/attachment-association-jobs/{created['job_id']}",
            headers=headers,
        ).json()

        assert result["status"] == "succeeded"
        assert result["resolution"] == "requires_confirmation"
        assert result["requires_confirmation"] is True
        assert len(result["candidates"]) == 2
        assert result["uploaded_count"] == 0
        receipt = ReceiptFolderService().get_receipt(receipt_id, current_user)
        assert receipt.status == "unlinked"
        with session_factory() as db:
            assert (
                db.scalar(select(BusinessEvent).where(BusinessEvent.aggregate_type == "receipt"))
                is None
            )
            assert (
                db.scalar(select(ExpenseCaseLink).where(ExpenseCaseLink.resource_type == "receipt"))
                is None
            )
            claims = list(
                db.scalars(select(ExpenseClaim).options(selectinload(ExpenseClaim.items))).all()
            )
            assert all(not item.invoice_id for claim in claims for item in claim.items)
    finally:
        clear_attachment_association_jobs_for_tests()
        get_settings.cache_clear()


def test_attachment_association_rolls_back_when_event_write_fails(monkeypatch, tmp_path) -> None:
    monkeypatch.setenv("STORAGE_ROOT_DIR", str(tmp_path / "storage"))
    get_settings.cache_clear()
    clear_attachment_association_jobs_for_tests()
    monkeypatch.setattr(OcrService, "recognize_files", fake_ocr_recognize)
    monkeypatch.setattr(
        ExpenseClaimAttachmentStorage, "root", lambda self: tmp_path / "attachments"
    )
    original_record = ExpenseCaseService.record_resource_event

    def fail_association_event(self, expense_case, **kwargs):
        if kwargs.get("event_type") == "attachment_associated":
            raise RuntimeError("simulated business event failure")
        return original_record(self, expense_case, **kwargs)

    monkeypatch.setattr(ExpenseCaseService, "record_resource_event", fail_association_event)
    try:
        client, session_factory = build_client(monkeypatch)
        current_user = CurrentUserContext(
            tenant_id="default",
            username="zhangsan@example.com",
            name="张三",
            role_codes=["user"],
            is_admin=False,
            employee_no="E10001",
        )
        with session_factory() as db:
            seed_travel_claim(db)
        receipt_id = save_train_receipt(
            service=ReceiptFolderService(),
            current_user=current_user,
            filename="2月20 武汉-上海.pdf",
            route="武汉-上海",
            trip_date="2026-02-20",
        )
        headers = {
            "x-auth-username": "zhangsan@example.com",
            "x-auth-name": "Zhang San",
            "x-auth-employee-no": "E10001",
            "x-auth-role-codes": "user",
        }
        created = client.post(
            "/api/v1/reimbursements/attachment-association-jobs",
            headers=headers,
            json={"receipt_ids": [receipt_id]},
        ).json()
        result = client.get(
            f"/api/v1/reimbursements/attachment-association-jobs/{created['job_id']}",
            headers=headers,
        ).json()

        assert result["status"] == "failed"
        assert "simulated business event failure" in result["error"]
        receipt = ReceiptFolderService().get_receipt(receipt_id, current_user)
        assert receipt.status == "unlinked"
        with session_factory() as db:
            claim = db.scalar(
                select(ExpenseClaim)
                .options(selectinload(ExpenseClaim.items))
                .where(ExpenseClaim.id == "claim-bg-association")
            )
            assert claim is not None
            assert all(not item.invoice_id for item in claim.items)
            assert (
                db.scalar(select(BusinessEvent).where(BusinessEvent.aggregate_type == "receipt"))
                is None
            )
    finally:
        clear_attachment_association_jobs_for_tests()
        get_settings.cache_clear()


def test_attachment_association_failure_keeps_preexisting_application_case(
    monkeypatch,
    tmp_path,
) -> None:
    monkeypatch.setenv("STORAGE_ROOT_DIR", str(tmp_path / "storage"))
    get_settings.cache_clear()
    clear_attachment_association_jobs_for_tests()
    monkeypatch.setattr(OcrService, "recognize_files", fake_ocr_recognize)
    monkeypatch.setattr(
        ExpenseClaimAttachmentStorage, "root", lambda self: tmp_path / "attachments"
    )
    original_record = ExpenseCaseService.record_resource_event

    def fail_association_event(self, expense_case, **kwargs):
        if kwargs.get("event_type") == "attachment_associated":
            raise RuntimeError("simulated application case event failure")
        return original_record(self, expense_case, **kwargs)

    monkeypatch.setattr(ExpenseCaseService, "record_resource_event", fail_association_event)
    try:
        client, session_factory = build_client(monkeypatch)
        current_user = CurrentUserContext(
            tenant_id="default",
            username="zhangsan@example.com",
            name="张三",
            role_codes=["user"],
            is_admin=False,
            employee_no="E10001",
        )
        with session_factory() as db:
            draft = seed_travel_claim(db)
            application, expense_case = seed_approved_application_with_flag_only(db, draft)
            application_id = application.id
            expense_case_id = expense_case.id
        receipt_id = save_train_receipt(
            service=ReceiptFolderService(),
            current_user=current_user,
            filename="2月20 武汉-上海.pdf",
            route="武汉-上海",
            trip_date="2026-02-20",
        )
        headers = {
            "x-auth-username": "zhangsan@example.com",
            "x-auth-name": "Zhang San",
            "x-auth-employee-no": "E10001",
            "x-auth-role-codes": "user",
        }
        created = client.post(
            "/api/v1/reimbursements/attachment-association-jobs",
            headers=headers,
            json={"receipt_ids": [receipt_id]},
        ).json()
        result = client.get(
            f"/api/v1/reimbursements/attachment-association-jobs/{created['job_id']}",
            headers=headers,
        ).json()

        assert result["status"] == "failed"
        with session_factory() as db:
            assert db.get(ExpenseCase, expense_case_id) is not None
            application_link = db.scalar(
                select(ExpenseCaseLink).where(
                    ExpenseCaseLink.resource_type == "expense_claim",
                    ExpenseCaseLink.resource_id == application_id,
                )
            )
            assert application_link is not None
            assert application_link.expense_case_id == expense_case_id
            assert (
                db.scalar(
                    select(ExpenseCaseLink).where(
                        ExpenseCaseLink.resource_type == "expense_claim",
                        ExpenseCaseLink.resource_id == "claim-bg-association",
                    )
                )
                is None
            )
    finally:
        clear_attachment_association_jobs_for_tests()
        get_settings.cache_clear()


def test_non_default_tenant_does_not_match_unscoped_legacy_claim(
    monkeypatch,
    tmp_path,
) -> None:
    monkeypatch.setenv("STORAGE_ROOT_DIR", str(tmp_path / "storage"))
    get_settings.cache_clear()
    clear_attachment_association_jobs_for_tests()
    try:
        client, session_factory = build_client(monkeypatch)
        with session_factory() as db:
            seed_travel_claim(db)
        tenant_user = CurrentUserContext(
            username="zhangsan@example.com",
            name="张三",
            role_codes=["user"],
            is_admin=False,
            tenant_id="tenant-b",
            employee_no="E10001",
        )
        receipt_id = save_train_receipt(
            service=ReceiptFolderService(),
            current_user=tenant_user,
            filename="tenant-b-ticket.pdf",
            route="武汉-上海",
            trip_date="2026-02-20",
        )
        headers = {
            "x-auth-username": "zhangsan@example.com",
            "x-auth-name": "Zhang San",
            "x-auth-tenant-id": "tenant-b",
            "x-auth-employee-no": "E10001",
            "x-auth-role-codes": "user",
        }
        created = client.post(
            "/api/v1/reimbursements/attachment-association-jobs",
            headers=headers,
            json={"receipt_ids": [receipt_id]},
        ).json()
        result = client.get(
            f"/api/v1/reimbursements/attachment-association-jobs/{created['job_id']}",
            headers=headers,
        ).json()

        assert result["status"] == "succeeded"
        assert result["resolution"] == "requires_confirmation"
        assert result["candidates"] == []
        assert result["uploaded_count"] == 0
        with session_factory() as db:
            claim = db.scalar(
                select(ExpenseClaim)
                .options(selectinload(ExpenseClaim.items))
                .where(ExpenseClaim.id == "claim-bg-association")
            )
            assert claim is not None
            assert all(not item.invoice_id for item in claim.items)
    finally:
        clear_attachment_association_jobs_for_tests()
        get_settings.cache_clear()


def test_confirmation_match_does_not_repair_or_commit_unrelated_claim(
    monkeypatch,
    tmp_path,
) -> None:
    monkeypatch.setenv("STORAGE_ROOT_DIR", str(tmp_path / "storage"))
    get_settings.cache_clear()
    clear_attachment_association_jobs_for_tests()
    try:
        client, session_factory = build_client(monkeypatch)
        with session_factory() as db:
            employee = Employee(
                id="emp-readonly-confirmation",
                employee_no="E10001",
                name="张三",
                email="zhangsan@example.com",
            )
            claim = ExpenseClaim(
                id="claim-readonly-confirmation",
                claim_no="BX-20260101-READONLY",
                employee_id=employee.id,
                employee_name=employee.name,
                department_name="交付部",
                expense_type="travel",
                reason="北京客户拜访",
                location="北京",
                amount=Decimal("100.00"),
                currency="CNY",
                invoice_count=0,
                occurred_at=datetime(2026, 1, 1, tzinfo=UTC),
                status="submitted",
                approval_stage="预算管理者审批",
                risk_flags_json=[
                    {
                        "source": "manual_approval",
                        "event_type": "expense_claim_approval",
                        "previous_approval_stage": "直属领导审批",
                        "next_approval_stage": "预算管理者审批",
                        "operator": "同一审批人",
                        "next_approver_name": "同一审批人",
                    }
                ],
            )
            db.add_all([employee, claim])
            db.commit()
        current_user = CurrentUserContext(
            tenant_id="default",
            username="zhangsan@example.com",
            name="张三",
            role_codes=["user"],
            is_admin=False,
            employee_no="E10001",
        )
        receipt_id = save_train_receipt(
            service=ReceiptFolderService(),
            current_user=current_user,
            filename="4月20 武汉-上海.pdf",
            route="武汉-上海",
            trip_date="2026-04-20",
        )
        headers = {
            "x-auth-username": "zhangsan@example.com",
            "x-auth-name": "Zhang San",
            "x-auth-employee-no": "E10001",
            "x-auth-role-codes": "user",
        }
        created = client.post(
            "/api/v1/reimbursements/attachment-association-jobs",
            headers=headers,
            json={"receipt_ids": [receipt_id]},
        ).json()
        result = client.get(
            f"/api/v1/reimbursements/attachment-association-jobs/{created['job_id']}",
            headers=headers,
        ).json()

        assert result["resolution"] == "requires_confirmation"
        with session_factory() as db:
            claim = db.get(ExpenseClaim, "claim-readonly-confirmation")
            assert claim is not None
            assert claim.approval_stage == "预算管理者审批"
            assert len(claim.risk_flags_json) == 1
            assert claim.risk_flags_json[0]["source"] == "manual_approval"
    finally:
        clear_attachment_association_jobs_for_tests()
        get_settings.cache_clear()


def test_mixed_receipt_batch_requires_confirmation_without_writes(
    monkeypatch,
    tmp_path,
) -> None:
    monkeypatch.setenv("STORAGE_ROOT_DIR", str(tmp_path / "storage"))
    get_settings.cache_clear()
    clear_attachment_association_jobs_for_tests()
    try:
        client, session_factory = build_client(monkeypatch)
        current_user = CurrentUserContext(
            tenant_id="default",
            username="zhangsan@example.com",
            name="张三",
            role_codes=["user"],
            is_admin=False,
            employee_no="E10001",
        )
        with session_factory() as db:
            draft = seed_travel_claim(db)
            seed_approved_application_for_draft(db, draft)
        matching_receipt_id = save_train_receipt(
            service=ReceiptFolderService(),
            current_user=current_user,
            filename="2月20 武汉-上海.pdf",
            route="武汉-上海",
            trip_date="2026-02-20",
        )
        unrelated_receipt_id = save_train_receipt(
            service=ReceiptFolderService(),
            current_user=current_user,
            filename="5月12 北京-广州.pdf",
            route="北京-广州",
            trip_date="2026-05-12",
        )
        headers = {
            "x-auth-username": "zhangsan@example.com",
            "x-auth-name": "Zhang San",
            "x-auth-employee-no": "E10001",
            "x-auth-role-codes": "user",
        }
        created = client.post(
            "/api/v1/reimbursements/attachment-association-jobs",
            headers=headers,
            json={"receipt_ids": [matching_receipt_id, unrelated_receipt_id]},
        ).json()
        result = client.get(
            f"/api/v1/reimbursements/attachment-association-jobs/{created['job_id']}",
            headers=headers,
        ).json()

        assert result["resolution"] == "requires_confirmation"
        assert "每一份" in result["exceptions"][0]
        assert result["uploaded_count"] == 0
        assert all(
            ReceiptFolderService().get_receipt(receipt_id, current_user).status == "unlinked"
            for receipt_id in (matching_receipt_id, unrelated_receipt_id)
        )
        with session_factory() as db:
            assert (
                db.scalar(select(BusinessEvent).where(BusinessEvent.aggregate_type == "receipt"))
                is None
            )
    finally:
        clear_attachment_association_jobs_for_tests()
        get_settings.cache_clear()


def test_attachment_write_failure_restores_previous_directory_and_business_state(
    monkeypatch,
    tmp_path,
) -> None:
    monkeypatch.setenv("STORAGE_ROOT_DIR", str(tmp_path / "storage"))
    get_settings.cache_clear()
    clear_attachment_association_jobs_for_tests()
    monkeypatch.setattr(OcrService, "recognize_files", fake_ocr_recognize)
    monkeypatch.setattr(
        ExpenseClaimAttachmentStorage, "root", lambda self: tmp_path / "attachments"
    )
    original_write_meta = ExpenseClaimAttachmentStorage.write_meta

    def write_meta_then_fail(self, file_path, payload):
        original_write_meta(self, file_path, payload)
        raise OSError("simulated attachment metadata failure")

    monkeypatch.setattr(ExpenseClaimAttachmentStorage, "write_meta", write_meta_then_fail)
    try:
        client, session_factory = build_client(monkeypatch)
        current_user = CurrentUserContext(
            tenant_id="default",
            username="zhangsan@example.com",
            name="张三",
            role_codes=["user"],
            is_admin=False,
            employee_no="E10001",
        )
        with session_factory() as db:
            draft = seed_travel_claim(db)
            item_id = draft.items[0].id
        item_dir = ExpenseClaimAttachmentStorage().build_item_dir(
            "claim-bg-association",
            item_id,
        )
        item_dir.mkdir(parents=True, exist_ok=True)
        sentinel = item_dir / "existing-state.txt"
        sentinel.write_text("keep-me", encoding="utf-8")
        receipt_id = save_train_receipt(
            service=ReceiptFolderService(),
            current_user=current_user,
            filename="2月20 武汉-上海.pdf",
            route="武汉-上海",
            trip_date="2026-02-20",
        )
        headers = {
            "x-auth-username": "zhangsan@example.com",
            "x-auth-name": "Zhang San",
            "x-auth-employee-no": "E10001",
            "x-auth-role-codes": "user",
        }
        created = client.post(
            "/api/v1/reimbursements/attachment-association-jobs",
            headers=headers,
            json={"receipt_ids": [receipt_id]},
        ).json()
        result = client.get(
            f"/api/v1/reimbursements/attachment-association-jobs/{created['job_id']}",
            headers=headers,
        ).json()

        assert result["status"] == "failed"
        assert "simulated attachment metadata failure" in result["error"]
        assert sentinel.read_text(encoding="utf-8") == "keep-me"
        assert [path.name for path in item_dir.iterdir()] == ["existing-state.txt"]
        assert ReceiptFolderService().get_receipt(receipt_id, current_user).status == "unlinked"
        with session_factory() as db:
            claim = db.scalar(
                select(ExpenseClaim)
                .options(selectinload(ExpenseClaim.items))
                .where(ExpenseClaim.id == "claim-bg-association")
            )
            assert claim is not None
            assert all(not item.invoice_id for item in claim.items)
            assert (
                db.scalar(select(BusinessEvent).where(BusinessEvent.aggregate_type == "receipt"))
                is None
            )
    finally:
        clear_attachment_association_jobs_for_tests()
        get_settings.cache_clear()


def test_concurrent_jobs_for_same_receipt_are_serialized(monkeypatch, tmp_path) -> None:
    monkeypatch.setenv("STORAGE_ROOT_DIR", str(tmp_path / "storage"))
    get_settings.cache_clear()
    clear_attachment_association_jobs_for_tests()
    monkeypatch.setattr(OcrService, "recognize_files", fake_ocr_recognize)
    monkeypatch.setattr(
        ExpenseClaimAttachmentStorage, "root", lambda self: tmp_path / "attachments"
    )
    try:
        _client, session_factory = build_client(monkeypatch)
        current_user = CurrentUserContext(
            tenant_id="default",
            username="zhangsan@example.com",
            name="张三",
            role_codes=["user"],
            is_admin=False,
            employee_no="E10001",
        )
        with session_factory() as db:
            seed_travel_claim(db)
        receipt_id = save_train_receipt(
            service=ReceiptFolderService(),
            current_user=current_user,
            filename="2月20 武汉-上海.pdf",
            route="武汉-上海",
            trip_date="2026-02-20",
        )
        with session_factory() as db:
            job_a = create_attachment_association_job(
                AttachmentAssociationJobCreate(receipt_ids=[receipt_id]),
                current_user,
                db,
            )
            job_b = create_attachment_association_job(
                AttachmentAssociationJobCreate(receipt_ids=[receipt_id]),
                current_user,
                db,
            )
        assert job_a.job_id == job_b.job_id
        original_associate = ExpenseReceiptAssociationService.associate
        active_lock = Lock()
        active_count = 0
        max_active_count = 0

        def tracked_associate(self, **kwargs):
            nonlocal active_count, max_active_count
            with active_lock:
                active_count += 1
                max_active_count = max(max_active_count, active_count)
            sleep(0.05)
            try:
                return original_associate(self, **kwargs)
            finally:
                with active_lock:
                    active_count -= 1

        monkeypatch.setattr(ExpenseReceiptAssociationService, "associate", tracked_associate)
        threads = [
            Thread(
                target=run_attachment_association_job,
                args=(job_id, session_factory),
            )
            for job_id in (job_a.job_id, job_b.job_id)
        ]
        for thread in threads:
            thread.start()
        for thread in threads:
            thread.join(timeout=5)
            assert not thread.is_alive()

        with session_factory() as db:
            result = get_attachment_association_job(job_a.job_id, current_user, db)
        assert max_active_count == 1
        assert result is not None and result.status == "succeeded"
        assert result.uploaded_count == 1
        assert result.skipped_count == 0
        assert ReceiptFolderService().get_receipt(receipt_id, current_user).status == "linked"
        with session_factory() as db:
            claim = db.scalar(
                select(ExpenseClaim)
                .options(selectinload(ExpenseClaim.items))
                .where(ExpenseClaim.id == "claim-bg-association")
            )
            assert claim is not None
            attached_items = [item for item in claim.items if item.invoice_id]
            assert len(attached_items) == 1
            attachment_path = ExpenseClaimAttachmentStorage().resolve_item_path(attached_items[0])
            assert attachment_path is not None and attachment_path.exists()
            events = list(
                db.scalars(
                    select(BusinessEvent).where(BusinessEvent.aggregate_type == "receipt")
                ).all()
            )
            assert len(events) == 2
    finally:
        clear_attachment_association_jobs_for_tests()
        get_settings.cache_clear()


def test_concurrent_receipts_for_same_claim_are_serialized(monkeypatch, tmp_path) -> None:
    monkeypatch.setenv("STORAGE_ROOT_DIR", str(tmp_path / "storage"))
    get_settings.cache_clear()
    clear_attachment_association_jobs_for_tests()
    monkeypatch.setattr(OcrService, "recognize_files", fake_ocr_recognize)
    monkeypatch.setattr(
        ExpenseClaimAttachmentStorage, "root", lambda self: tmp_path / "attachments"
    )
    try:
        _client, session_factory = build_client(monkeypatch)
        current_user = CurrentUserContext(
            tenant_id="default",
            username="zhangsan@example.com",
            name="张三",
            role_codes=["user"],
            is_admin=False,
            employee_no="E10001",
        )
        with session_factory() as db:
            seed_travel_claim(db)
        receipt_ids = [
            save_train_receipt(
                service=ReceiptFolderService(),
                current_user=current_user,
                filename=filename,
                route="武汉-上海",
                trip_date="2026-02-20",
            )
            for filename in ("并发票据-A.pdf", "并发票据-B.pdf")
        ]
        with session_factory() as db:
            jobs = [
                create_attachment_association_job(
                    AttachmentAssociationJobCreate(receipt_ids=[receipt_id]),
                    current_user,
                    db,
                )
                for receipt_id in receipt_ids
            ]
        assert jobs[0].job_id != jobs[1].job_id

        original_associate_matched = ExpenseReceiptAssociationService._associate_matched
        active_lock = Lock()
        active_count = 0
        max_active_count = 0

        def tracked_associate_matched(self, **kwargs):
            nonlocal active_count, max_active_count
            with active_lock:
                active_count += 1
                max_active_count = max(max_active_count, active_count)
            sleep(0.05)
            try:
                return original_associate_matched(self, **kwargs)
            finally:
                with active_lock:
                    active_count -= 1

        monkeypatch.setattr(
            ExpenseReceiptAssociationService,
            "_associate_matched",
            tracked_associate_matched,
        )
        threads = [
            Thread(
                target=run_attachment_association_job,
                args=(job.job_id, session_factory),
            )
            for job in jobs
        ]
        for thread in threads:
            thread.start()
        for thread in threads:
            thread.join(timeout=5)
            assert not thread.is_alive()

        with session_factory() as db:
            results = [get_attachment_association_job(job.job_id, current_user, db) for job in jobs]
            claim = db.scalar(
                select(ExpenseClaim)
                .options(selectinload(ExpenseClaim.items))
                .where(ExpenseClaim.id == "claim-bg-association")
            )
            events = list(
                db.scalars(
                    select(BusinessEvent).where(BusinessEvent.aggregate_type == "receipt")
                ).all()
            )
            receipt_links = list(
                db.scalars(
                    select(ExpenseCaseLink).where(ExpenseCaseLink.resource_type == "receipt")
                ).all()
            )

            assert max_active_count == 1
            assert all(result is not None and result.status == "succeeded" for result in results)
            assert all(result.resolution == "auto_associated" for result in results if result)
            assert all(result.uploaded_count == 1 for result in results if result)
            assert claim is not None
            attached_items = [item for item in claim.items if item.invoice_id]
            assert len(attached_items) == 2
            assert claim.invoice_count == 2
            assert len({item.invoice_id for item in attached_items}) == 2
            assert all(
                (path := ExpenseClaimAttachmentStorage().resolve_item_path(item)) is not None
                and path.exists()
                for item in attached_items
            )
            assert len(events) == 4
            assert len(receipt_links) == 2
            assert {(event.aggregate_id, event.event_type) for event in events} == {
                (receipt_id, event_type)
                for receipt_id in receipt_ids
                for event_type in ("receipt_received", "attachment_associated")
            }
        linked_receipts = [
            ReceiptFolderService().get_receipt(receipt_id, current_user)
            for receipt_id in receipt_ids
        ]
        assert all(receipt.status == "linked" for receipt in linked_receipts)
        assert all(receipt.linked_claim_id == "claim-bg-association" for receipt in linked_receipts)
        assert (
            len(
                {
                    str((receipt.raw_meta or {}).get("linked_item_id") or "")
                    for receipt in linked_receipts
                }
            )
            == 2
        )
    finally:
        clear_attachment_association_jobs_for_tests()
        get_settings.cache_clear()


def test_persistent_job_resumes_after_process_state_is_cleared(monkeypatch, tmp_path) -> None:
    monkeypatch.setenv("STORAGE_ROOT_DIR", str(tmp_path / "storage"))
    get_settings.cache_clear()
    clear_attachment_association_jobs_for_tests()
    monkeypatch.setattr(OcrService, "recognize_files", fake_ocr_recognize)
    monkeypatch.setattr(
        ExpenseClaimAttachmentStorage, "root", lambda self: tmp_path / "attachments"
    )
    try:
        client, session_factory = build_client(monkeypatch)
        current_user = CurrentUserContext(
            tenant_id="default",
            username="zhangsan@example.com",
            name="张三",
            role_codes=["user"],
            is_admin=False,
            employee_no="E10001",
        )
        with session_factory() as db:
            seed_travel_claim(db)
        receipt_id = save_train_receipt(
            service=ReceiptFolderService(),
            current_user=current_user,
            filename="2月20 武汉-上海.pdf",
            route="武汉-上海",
            trip_date="2026-02-20",
        )
        with session_factory() as db:
            queued_job = create_attachment_association_job(
                AttachmentAssociationJobCreate(receipt_ids=[receipt_id]),
                current_user,
                db,
            )
            persisted_job = db.get(AttachmentAssociationJob, queued_job.job_id)
            assert persisted_job is not None
            persisted_job.status = "running"
            persisted_job.attempt_count = 1
            persisted_job.lease_expires_at = datetime.now(UTC) - timedelta(seconds=1)
            db.commit()
        clear_attachment_association_jobs_for_tests()
        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.get(
            f"/api/v1/reimbursements/attachment-association-jobs/{queued_job.job_id}",
            headers=headers,
        )
        assert first_response.status_code == 200
        assert first_response.json()["status"] == "running"
        second_response = client.get(
            f"/api/v1/reimbursements/attachment-association-jobs/{queued_job.job_id}",
            headers=headers,
        )
        assert second_response.status_code == 200
        assert second_response.json()["status"] == "succeeded"
        assert second_response.json()["uploaded_count"] == 1
    finally:
        clear_attachment_association_jobs_for_tests()
        get_settings.cache_clear()


def test_expired_worker_cannot_overwrite_newer_job_attempt(monkeypatch) -> None:
    clear_attachment_association_jobs_for_tests()
    try:
        _client, session_factory = build_client(monkeypatch)
        current_user = CurrentUserContext(
            tenant_id="default",
            username="zhangsan@example.com",
            name="张三",
            role_codes=["user"],
            is_admin=False,
            employee_no="E10001",
        )
        with session_factory() as db:
            created = create_attachment_association_job(
                AttachmentAssociationJobCreate(receipt_ids=["lease-fencing-receipt"]),
                current_user,
                db,
            )
            first_attempt = claim_persistent_job(db, created.job_id)
            assert first_attempt is not None and first_attempt.attempt_count == 1
            job = db.get(AttachmentAssociationJob, created.job_id)
            assert job is not None
            job.lease_expires_at = datetime.now(UTC) - timedelta(seconds=1)
            db.commit()

        with session_factory() as db:
            second_attempt = claim_persistent_job(db, created.job_id)
            assert second_attempt is not None and second_attempt.attempt_count == 2

        with session_factory() as db:
            update_persistent_job(
                db,
                created.job_id,
                expected_attempt_count=1,
                status="failed",
                resolution="failed",
                error="stale worker must not win",
            )
            current = db.get(AttachmentAssociationJob, created.job_id)
            assert current is not None
            assert current.status == "running"
            assert current.attempt_count == 2
            assert current.error == ""
            assert current.lease_expires_at is not None

        with session_factory() as db:
            update_persistent_job(
                db,
                created.job_id,
                expected_attempt_count=2,
                status="succeeded",
                resolution="auto_associated",
                message="new worker succeeded",
                error="",
            )
            current = db.get(AttachmentAssociationJob, created.job_id)
            assert current is not None
            assert current.status == "succeeded"
            assert current.attempt_count == 2
            assert current.lease_expires_at is None
            assert current.message == "new worker succeeded"

        with session_factory() as db:
            update_persistent_job(
                db,
                created.job_id,
                expected_attempt_count=2,
                status="failed",
                resolution="failed",
                message="late callback",
                error="late callback",
            )
            current = db.get(AttachmentAssociationJob, created.job_id)
            assert current is not None
            assert current.status == "succeeded"
            assert current.resolution == "auto_associated"
            assert current.message == "new worker succeeded"
            assert current.error == ""
    finally:
        clear_attachment_association_jobs_for_tests()


def test_failed_job_creates_new_generation_without_rewriting_history(monkeypatch) -> None:
    clear_attachment_association_jobs_for_tests()
    try:
        _client, session_factory = build_client(monkeypatch)
        current_user = CurrentUserContext(
            tenant_id="default",
            username="zhangsan@example.com",
            name="张三",
            role_codes=["user"],
            is_admin=False,
            employee_no="E10001",
        )
        payload = AttachmentAssociationJobCreate(receipt_ids=["failed-generation-receipt"])
        with session_factory() as db:
            first = create_attachment_association_job(payload, current_user, db)
            claimed = claim_persistent_job(db, first.job_id)
            assert claimed is not None
            update_persistent_job(
                db,
                first.job_id,
                expected_attempt_count=claimed.attempt_count,
                status="failed",
                resolution="failed",
                message="generation one failed",
                error="generation one failed",
            )

        with session_factory() as db:
            second = create_attachment_association_job(payload, current_user, db)
            jobs = list(
                db.scalars(
                    select(AttachmentAssociationJob).order_by(AttachmentAssociationJob.generation)
                ).all()
            )

        assert second.job_id != first.job_id
        assert second.status == "queued"
        assert [(job.generation, job.status) for job in jobs] == [
            (1, "failed"),
            (2, "queued"),
        ]
        assert jobs[0].message == "generation one failed"
        assert jobs[0].error == "generation one failed"
    finally:
        clear_attachment_association_jobs_for_tests()


def test_attachment_association_job_and_receipts_are_tenant_scoped(monkeypatch, tmp_path) -> None:
    monkeypatch.setenv("STORAGE_ROOT_DIR", str(tmp_path / "storage"))
    get_settings.cache_clear()
    clear_attachment_association_jobs_for_tests()
    try:
        client, _session_factory = build_client(monkeypatch)
        tenant_a_user = CurrentUserContext(
            username="shared@example.com",
            name="同名用户",
            role_codes=["user"],
            is_admin=False,
            tenant_id="tenant-a",
            employee_no="EA001",
        )
        receipt_id = save_train_receipt(
            service=ReceiptFolderService(),
            current_user=tenant_a_user,
            filename="tenant-a-ticket.pdf",
            route="武汉-上海",
            trip_date="2026-02-20",
        )
        tenant_a_headers = {
            "x-auth-username": "shared@example.com",
            "x-auth-name": "Same User",
            "x-auth-tenant-id": "tenant-a",
            "x-auth-employee-no": "EA001",
            "x-auth-role-codes": "user",
        }
        tenant_b_headers = {**tenant_a_headers, "x-auth-tenant-id": "tenant-b"}
        created = client.post(
            "/api/v1/reimbursements/attachment-association-jobs",
            headers=tenant_a_headers,
            json={"receipt_ids": [receipt_id]},
        )
        assert created.status_code == 202
        job_id = created.json()["job_id"]
        assert (
            client.get(
                f"/api/v1/reimbursements/attachment-association-jobs/{job_id}",
                headers=tenant_b_headers,
            ).status_code
            == 404
        )
        assert (
            client.get(
                f"/api/v1/reimbursements/attachment-association-jobs/{job_id}",
                headers={**tenant_b_headers, "x-auth-is-admin": "true"},
            ).status_code
            == 404
        )

        tenant_b_user = CurrentUserContext(
            username="shared@example.com",
            name="同名用户",
            role_codes=["user"],
            is_admin=False,
            tenant_id="tenant-b",
            employee_no="EA001",
        )
        try:
            ReceiptFolderService().get_receipt(receipt_id, tenant_b_user)
        except FileNotFoundError:
            pass
        else:
            raise AssertionError("跨租户不应读取同用户名票据")

        collision_a_user = CurrentUserContext(
            username="collision@example.com",
            name="命名空间碰撞用户",
            role_codes=["user"],
            is_admin=False,
            tenant_id="tenant/a",
            employee_no="EA002",
        )
        collision_receipt_id = save_train_receipt(
            service=ReceiptFolderService(),
            current_user=collision_a_user,
            filename="collision-ticket.pdf",
            route="武汉-上海",
            trip_date="2026-02-20",
        )
        collision_b_user = CurrentUserContext(
            username="collision@example.com",
            name="命名空间碰撞用户",
            role_codes=["user"],
            is_admin=False,
            tenant_id="tenant?a",
            employee_no="EA002",
        )
        try:
            ReceiptFolderService().get_receipt(collision_receipt_id, collision_b_user)
        except FileNotFoundError:
            pass
        else:
            raise AssertionError("规范化后同名的不同租户也不应共享票据目录")
    finally:
        clear_attachment_association_jobs_for_tests()
        get_settings.cache_clear()