YG-Soft/X-Financial
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ai): add expense application feedback ledger

Browse Source
This commit is contained in:
caoxiaozhu
2026-07-14 11:10:55 +08:00
parent 5ed34c2b8f
commit a662cfe6c3
24 changed files with 2159 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

179
server/tests/test_user_agent_application_draft_events.py
View File

@@ -9,6 +9,7 @@ from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool
from app.db.base import Base
from app.models.ai_learning import AIDecision, AIDecisionFeedback, WorkflowOutcome
from app.models.employee import Employee
from app.models.expense_case import BusinessEvent, ExpenseCaseLink
from app.models.financial_record import ExpenseClaim
@@ -28,7 +29,12 @@ def build_session_factory() -> sessionmaker[Session]:
return sessionmaker(bind=engine, autoflush=False, autocommit=False)
def build_request(*, run_id: str, tenant_id: str = "tenant-a") -> UserAgentRequest:
def build_request(
*,
run_id: str,
tenant_id: str = "tenant-a",
changed_fields: list[dict[str, str]] | None = None,
) -> UserAgentRequest:
return UserAgentRequest(
run_id=run_id,
user_id="owner@example.com",
@@ -40,6 +46,20 @@ def build_request(*, run_id: str, tenant_id: str = "tenant-a") -> UserAgentReque
"name": "张三",
"employee_no": "E001",
"department_name": "市场部",
"application_preview": {
"modelReviewStatus": "completed",
"fields": {
"applicationType": "差旅费用申请",
"time": "2026-07-13 至 2026-07-14",
"location": "上海",
"reason": "客户现场差旅",
"amount": "880元",
},
"aiDecisionFeedback": {
"schemaVersion": 1,
"changedFields": changed_fields or [],
},
},
},
)
@@ -84,11 +104,14 @@ def test_ai_application_draft_update_writes_tenant_scoped_event() -> None:
db.add_all([owner, claim])
db.commit()
updated = UserAgentService(db)._update_expense_application_record(
build_request(run_id="application-draft-update"),
service = UserAgentService(db)
request = build_request(run_id="application-draft-update")
updated = service._update_expense_application_record(
request,
build_facts(reason="更新后的申请事由"),
claim,
submit=False,
learning_current_user=service._build_application_current_user(request),
)
assert updated.status == "draft"
@@ -103,6 +126,65 @@ def test_ai_application_draft_update_writes_tenant_scoped_event() -> None:
link = db.scalar(select(ExpenseCaseLink).where(ExpenseCaseLink.resource_id == claim.id))
assert link is not None
assert link.tenant_id == "tenant-a"
decision = db.scalar(select(AIDecision).where(AIDecision.expense_claim_id == claim.id))
feedback = db.scalar(
select(AIDecisionFeedback).where(AIDecisionFeedback.expense_claim_id == claim.id)
)
outcome = db.scalar(
select(WorkflowOutcome).where(WorkflowOutcome.expense_claim_id == claim.id)
)
assert decision is not None
assert decision.business_event_id == event.id
assert feedback is not None
assert feedback.feedback_type == "accepted"
assert outcome is not None
assert outcome.outcome_type == "draft_saved"
def test_untrusted_application_path_does_not_write_learning_ledger() -> None:
session_factory = build_session_factory()
with session_factory() as db:
claim = UserAgentService(db)._create_expense_application_record(
build_request(run_id="application-untrusted-path"),
build_facts(),
submit=False,
)
assert db.get(ExpenseClaim, claim.id) is not None
assert db.scalar(
select(BusinessEvent).where(BusinessEvent.aggregate_id == claim.id)
) is not None
assert list(db.scalars(select(AIDecision)).all()) == []
assert list(db.scalars(select(AIDecisionFeedback)).all()) == []
assert list(db.scalars(select(WorkflowOutcome)).all()) == []
@pytest.mark.parametrize(
"preview_patch",
[
{"modelReviewStatus": "template"},
{"applicationEditMode": True},
],
)
def test_non_ai_or_detail_preview_is_not_recorded_as_ai_feedback(
preview_patch: dict[str, object],
) -> None:
session_factory = build_session_factory()
with session_factory() as db:
request = build_request(run_id="application-ineligible-preview")
request.context_json["application_preview"].update(preview_patch)
service = UserAgentService(db)
claim = service._create_expense_application_record(
request,
build_facts(),
submit=False,
learning_current_user=service._build_application_current_user(request),
)
assert db.get(ExpenseClaim, claim.id) is not None
assert list(db.scalars(select(AIDecision)).all()) == []
assert list(db.scalars(select(AIDecisionFeedback)).all()) == []
assert list(db.scalars(select(WorkflowOutcome)).all()) == []
def test_ai_application_draft_creation_rolls_back_when_event_write_fails(
@@ -217,8 +299,21 @@ def test_ai_application_draft_update_deduplicates_identical_snapshot() -> None:
request = build_request(run_id="application-draft-idempotent")
facts = build_facts(reason="同一份草稿")
service = UserAgentService(db)
service._update_expense_application_record(request, facts, claim, submit=False)
service._update_expense_application_record(request, facts, claim, submit=False)
learning_current_user = service._build_application_current_user(request)
service._update_expense_application_record(
request,
facts,
claim,
submit=False,
learning_current_user=learning_current_user,
)
service._update_expense_application_record(
request,
facts,
claim,
submit=False,
learning_current_user=learning_current_user,
)
events = list(
db.scalars(
@@ -228,6 +323,9 @@ def test_ai_application_draft_update_deduplicates_identical_snapshot() -> None:
assert len(events) == 1
assert events[0].event_type == "claim_draft_updated"
assert events[0].payload_json["previous_status"] == "returned"
assert len(list(db.scalars(select(AIDecision)).all())) == 1
assert len(list(db.scalars(select(AIDecisionFeedback)).all())) == 1
assert len(list(db.scalars(select(WorkflowOutcome)).all())) == 1
def test_ai_application_draft_update_keeps_distinct_snapshots_in_same_run() -> None:
@@ -262,17 +360,20 @@ def test_ai_application_draft_update_keeps_distinct_snapshots_in_same_run() -> N
request = build_request(run_id="application-draft-multi-version")
service = UserAgentService(db)
learning_current_user = service._build_application_current_user(request)
service._update_expense_application_record(
request,
build_facts(reason="第一版草稿"),
claim,
submit=False,
learning_current_user=learning_current_user,
)
service._update_expense_application_record(
request,
build_facts(reason="第二版草稿"),
claim,
submit=False,
learning_current_user=learning_current_user,
)
events = list(
@@ -283,3 +384,71 @@ def test_ai_application_draft_update_keeps_distinct_snapshots_in_same_run() -> N
assert len(events) == 2
assert {event.event_type for event in events} == {"claim_draft_updated"}
assert len({event.idempotency_key for event in events}) == 2
assert len(list(db.scalars(select(AIDecision)).all())) == 2
def test_ai_application_draft_records_explicit_field_correction() -> None:
session_factory = build_session_factory()
with session_factory() as db:
owner = Employee(
id="owner-1",
employee_no="E001",
name="张三",
email="owner@example.com",
)
claim = ExpenseClaim(
id="application-1",
claim_no="AP-DRAFT-FEEDBACK-001",
employee_id=owner.id,
employee_name=owner.name,
department_name="市场部",
expense_type="travel_application",
reason="原申请事由",
location="北京",
amount=Decimal("500.00"),
currency="CNY",
invoice_count=0,
occurred_at=datetime(2026, 7, 10, tzinfo=UTC),
submitted_at=None,
status="returned",
approval_stage="退回补充",
risk_flags_json=[],
)
db.add_all([owner, claim])
db.commit()
service = UserAgentService(db)
request = build_request(
run_id="application-draft-feedback",
changed_fields=[
{
"fieldKey": "reason",
"suggestedValue": "客户拜访",
"finalValue": "客户端伪造值不会直接入账",
}
],
)
service._update_expense_application_record(
request,
build_facts(reason="客户现场实施"),
claim,
submit=False,
learning_current_user=service._build_application_current_user(request),
)
decision = db.scalar(select(AIDecision))
feedback = db.scalar(select(AIDecisionFeedback))
assert decision is not None
assert decision.training_eligible is False
assert decision.suggestion_json["field_keys"]
assert decision.suggestion_json["value_fingerprint"].startswith("sha256:")
assert feedback is not None
assert feedback.feedback_type == "edited"
assert feedback.verification_status == "client_observed"
assert feedback.training_eligible is False
assert feedback.changed_fields_json[0]["field_key"] == "reason"
assert feedback.changed_fields_json[0]["suggested_value_fingerprint"].startswith(
"sha256:"
)
assert "客户拜访" not in str(feedback.changed_fields_json)
assert "客户现场实施" not in str(feedback.final_value_json)