from __future__ import annotations import shutil import uuid from io import BytesIO from pathlib import Path import pytest from openpyxl import Workbook, load_workbook from sqlalchemy import create_engine, select from sqlalchemy.orm import Session, sessionmaker from sqlalchemy.pool import StaticPool from app.api.deps import CurrentUserContext from app.core.agent_enums import ( AgentAssetContentType, AgentAssetDomain, AgentAssetStatus, AgentAssetType, AgentName, AgentReviewStatus, AgentRunSource, AgentRunStatus, ) from app.core.config import SERVER_DIR from app.db.base import Base from app.models.agent_asset import AgentAsset from app.models.employee import Employee from app.schemas.agent_asset import ( AgentAssetCreate, AgentAssetReviewCreate, AgentAssetVersionCreate, ) from app.schemas.reimbursement import TravelReimbursementCalculatorRequest from app.services import agent_foundation as agent_foundation_module from app.services.agent_asset_spreadsheet import ( COMPANY_COMMUNICATION_EXPENSE_RULE_CODE, COMPANY_COMMUNICATION_EXPENSE_RULE_FILENAME, COMPANY_TRAVEL_EXPENSE_RULE_CODE, COMPANY_TRAVEL_EXPENSE_RULE_FILENAME, FINANCE_RULES_LIBRARY, ) from app.services.agent_assets import AgentAssetService from app.services.agent_runs import AgentRunService from app.services.audit import AuditLogService from app.services.expense_rule_runtime import ExpenseRuleRuntimeService from app.services.finance_rule_catalog import ( DEPRECATED_FINANCE_RULE_CODES, ) from app.services.settings import OnlyOfficeRuntimeConfig from app.services.travel_reimbursement_calculator import TravelReimbursementCalculatorService @pytest.fixture(autouse=True) def isolate_rule_file_storage(tmp_path, monkeypatch) -> None: temp_server_dir = tmp_path / "server" temp_rules_root = temp_server_dir / "rules" temp_finance_rules = temp_rules_root / FINANCE_RULES_LIBRARY temp_finance_rules.mkdir(parents=True, exist_ok=True) real_finance_rules = SERVER_DIR / "rules" / FINANCE_RULES_LIBRARY for file_name in ( COMPANY_TRAVEL_EXPENSE_RULE_FILENAME, COMPANY_COMMUNICATION_EXPENSE_RULE_FILENAME, ): source_path = real_finance_rules / file_name if source_path.exists(): shutil.copy2(source_path, temp_finance_rules / file_name) monkeypatch.setattr( "app.services.agent_asset_spreadsheet.SERVER_DIR", temp_server_dir, ) def init_manager(self, storage_root=None, rule_root=None) -> None: self.storage_root = Path(storage_root or tmp_path / "storage").resolve() self.asset_root = (self.storage_root / "agent_assets").resolve() self.rule_root = Path(rule_root or temp_rules_root).resolve() monkeypatch.setattr( "app.services.agent_asset_spreadsheet.AgentAssetSpreadsheetManager.__init__", init_manager, ) def build_session() -> Session: agent_foundation_module._foundation_ready_keys.clear() engine = create_engine( "sqlite+pysqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool, ) Base.metadata.create_all(bind=engine) session_factory = sessionmaker(bind=engine, autoflush=False, autocommit=False) return session_factory() def build_workbook_bytes(rows: list[list[object]], *, sheet_name: str = "规则表") -> bytes: workbook = Workbook() sheet = workbook.active sheet.title = sheet_name for row in rows: sheet.append(row) buffer = BytesIO() workbook.save(buffer) return buffer.getvalue() def build_multi_sheet_workbook_bytes(sheets: dict[str, list[list[object]]]) -> bytes: workbook = Workbook() default_sheet = workbook.active workbook.remove(default_sheet) for sheet_name, rows in sheets.items(): sheet = workbook.create_sheet(sheet_name) for row in rows: sheet.append(row) buffer = BytesIO() workbook.save(buffer) return buffer.getvalue() def test_agent_asset_service_seeds_assets_and_enforces_review_before_activation() -> None: with build_session() as db: service = AgentAssetService(db) rules = service.list_assets(asset_type=AgentAssetType.RULE.value) assert len(rules) >= 3 assert any( item.code == "rule.expense.travel_risk_control_standard" and item.status == AgentAssetStatus.ACTIVE.value for item in rules ) assert all( item.code not in { "rule.expense.duplicate_expense_check", "rule.expense.travel_receipt_requirements", "rule.ap.payment_dual_review", } for item in rules ) pending_rule = next(item for item in rules if item.status == AgentAssetStatus.REVIEW.value) with pytest.raises(PermissionError): service.activate_asset(pending_rule.id, actor="pytest") def test_agent_asset_service_seeds_all_foundation_asset_types() -> None: with build_session() as db: service = AgentAssetService(db) assert len(service.list_assets(asset_type=AgentAssetType.RULE.value)) >= 3 assert len(service.list_assets(asset_type=AgentAssetType.SKILL.value)) >= 2 assert len(service.list_assets(asset_type=AgentAssetType.MCP.value)) >= 2 assert len(service.list_assets(asset_type=AgentAssetType.TASK.value)) >= 3 def test_agent_asset_service_supports_backend_pagination() -> None: with build_session() as db: service = AgentAssetService(db) page = service.list_assets_page( asset_type=AgentAssetType.RULE.value, page=1, page_size=2, ) assert len(page.items) <= 2 assert page.total >= len(page.items) assert page.page == 1 assert page.page_size == 2 def test_finance_rules_use_risk_rule_scenario_categories() -> None: with build_session() as db: service = AgentAssetService(db) rules = service.list_assets(asset_type=AgentAssetType.RULE.value) travel_rule = next(item for item in rules if item.code == COMPANY_TRAVEL_EXPENSE_RULE_CODE) communication_rule = next( item for item in rules if item.code == COMPANY_COMMUNICATION_EXPENSE_RULE_CODE ) travel_config = travel_rule.config_json or {} communication_config = communication_rule.config_json or {} assert travel_rule.scenario_json == ["差旅费"] assert travel_config["scenario_category"] == "差旅费" assert travel_config["ai_review_category"] == "差旅费" assert communication_rule.scenario_json == ["通信费"] assert communication_config["scenario_category"] == "通信费" assert communication_config["ai_review_category"] == "通信费" def test_non_standard_finance_rule_spreadsheets_are_not_seeded() -> None: with build_session() as db: service = AgentAssetService(db) service.list_assets(asset_type=AgentAssetType.RULE.value) for code in DEPRECATED_FINANCE_RULE_CODES: asset = db.scalar(select(AgentAsset).where(AgentAsset.code == code)) assert asset is None or asset.config_json["tag"] == "废弃规则" def test_demo_budget_risk_rules_sync_with_finance_rule_references() -> None: with build_session() as db: service = AgentAssetService(db) service.list_assets(asset_type=AgentAssetType.RULE.value) budget_rule = db.scalar( select(AgentAsset).where( AgentAsset.code == "risk.budget.available_balance_insufficient" ) ) marketing_rule = db.scalar( select(AgentAsset).where( AgentAsset.code == "risk.application.marketing_without_campaign" ) ) assert budget_rule is not None assert "差旅费" in budget_rule.scenario_json assert "市场推广费" in budget_rule.scenario_json assert "软件服务费" in budget_rule.scenario_json assert budget_rule.config_json["budget_required"] is True assert "marketing" in budget_rule.config_json["expense_types"] assert budget_rule.config_json["business_stage"] == [ "expense_application", "reimbursement", "budget_execution", ] assert budget_rule.config_json["finance_rule_code"] == "budget.execution.policy" assert marketing_rule is not None assert marketing_rule.scenario_json == ["市场推广费"] assert marketing_rule.config_json["finance_rule_code"] == "expense.application.policy" assert marketing_rule.config_json["finance_rule_sheet"] == "费用申请前置规则" assert marketing_rule.config_json["expense_types"] == ["marketing"] assert marketing_rule.config_json["budget_required"] is True def test_agent_asset_service_can_activate_rule_after_review() -> None: with build_session() as db: service = AgentAssetService(db) created = service.create_asset( AgentAssetCreate( asset_type=AgentAssetType.RULE, code=f"rule.test.{uuid.uuid4().hex[:8]}", name="测试规则", description="用于测试审核和上线流程。", domain=AgentAssetDomain.EXPENSE, scenario_json=["expense", "risk_check"], owner="pytest", reviewer="reviewer", status=AgentAssetStatus.DRAFT, config_json={"enabled": False}, ), actor="pytest", ) service.create_version( created.id, AgentAssetVersionCreate( version="v1.0.0", content="# 测试规则\n\n- 仅用于测试。", content_type=AgentAssetContentType.MARKDOWN, change_note="初始化版本", created_by="pytest", ), actor="pytest", ) service.create_review( created.id, AgentAssetReviewCreate( version="v1.0.0", reviewer="reviewer", review_status=AgentReviewStatus.APPROVED, review_note="可以上线", ), actor="reviewer", ) activated = service.activate_asset(created.id, actor="reviewer") assert activated.status == AgentAssetStatus.ACTIVE.value assert activated.current_version == "v1.0.0" assert activated.working_version == "v1.0.0" assert activated.published_version == "v1.0.0" assert activated.latest_review is not None assert activated.latest_review.review_status == AgentReviewStatus.APPROVED.value def test_rule_working_version_does_not_replace_published_version_until_activation() -> None: with build_session() as db: service = AgentAssetService(db) rule = next( item for item in service.list_assets(asset_type=AgentAssetType.RULE.value) if item.code == "rule.expense.travel_risk_control_standard" ) created = service.create_version( rule.id, AgentAssetVersionCreate( version="v1.1.1", content="# 差旅报销风险管控制度\n\n- 工作稿", content_type=AgentAssetContentType.MARKDOWN, change_note="新增工作稿", created_by="finance_user", ), actor="finance_user", ) detail = service.get_asset(rule.id) assert created.is_working is True assert created.is_published is False assert created.lifecycle_state == "draft" assert detail is not None assert detail.status == AgentAssetStatus.ACTIVE.value assert detail.current_version == "v1.1.1" assert detail.working_version == "v1.1.1" assert detail.published_version == "v1.1.0" assert detail.latest_review is None def test_pending_review_can_name_new_working_version_before_submission() -> None: with build_session() as db: service = AgentAssetService(db) rule = next( item for item in service.list_assets(asset_type=AgentAssetType.RULE.value) if item.code == "rule.expense.travel_risk_control_standard" ) review = service.create_review( rule.id, AgentAssetReviewCreate( version="v1.2.0", reviewer="manager_user", review_status=AgentReviewStatus.PENDING, review_note="请审核", ), actor="finance_user", ) detail = service.get_asset(rule.id) assert review.version == "v1.2.0" assert detail is not None assert detail.current_version == "v1.2.0" assert detail.working_version == "v1.2.0" assert detail.published_version == "v1.1.0" assert detail.latest_review is not None assert detail.latest_review.reviewer == "manager_user" def test_expense_rule_runtime_uses_published_version_instead_of_working_version() -> None: with build_session() as db: service = AgentAssetService(db) rule = next( item for item in service.list_assets(asset_type=AgentAssetType.RULE.value) if item.code == "rule.expense.travel_risk_control_standard" ) service.create_version( rule.id, AgentAssetVersionCreate( version="v1.1.1", content=( "# 工作稿\n\n" '```expense-rule\n{"kind":"travel_policy","version":1}\n```' ), content_type=AgentAssetContentType.MARKDOWN, change_note="未上线草稿", created_by="finance_user", ), actor="finance_user", ) catalog = ExpenseRuleRuntimeService(db).load_catalog() assert catalog.travel_policy is not None assert catalog.travel_policy.rule_version == "v1.1.0" def test_restore_version_creates_new_working_copy_without_rewriting_published_version() -> None: with build_session() as db: service = AgentAssetService(db) rule = next( item for item in service.list_assets(asset_type=AgentAssetType.RULE.value) if item.code == "rule.expense.travel_risk_control_standard" ) restored = service.restore_version_as_working_copy( rule.id, "v1.0.0", actor="manager_user", ) assert restored.working_version == "v1.1.1" assert restored.current_version == "v1.1.1" assert restored.published_version == "v1.1.0" assert restored.current_version_change_note == "基于历史版本 v1.0.0 恢复生成工作稿" def test_spreadsheet_upload_records_sheet_and_cell_changes_without_versions() -> None: with build_session() as db: service = AgentAssetService(db) rule = next( item for item in service.list_assets(asset_type=AgentAssetType.RULE.value) if item.code == "rule.expense.company_travel_expense_reimbursement" ) service.upload_rule_spreadsheet( rule.id, filename="公司差旅费报销规则.xlsx", content=build_workbook_bytes([["城市", "住宿"], ["北京", 500]]), actor="finance_user", ) service.upload_rule_spreadsheet( rule.id, filename="公司差旅费报销规则.xlsx", content=build_workbook_bytes([["城市", "住宿"], ["北京", 550], ["武汉", 450]]), actor="finance_user", ) records = service.list_spreadsheet_change_records(rule.id) latest = records[0] assert latest.changed_sheet_count == 1 assert latest.changed_cell_count == 3 assert any( item.cell == "B2" and item.change_type == "modified" for item in latest.cell_changes ) assert any( item.cell == "A3" and item.change_type == "added" for item in latest.cell_changes ) assert not hasattr(latest, "version") def test_spreadsheet_content_reads_current_rule_file_without_version_snapshot() -> None: with build_session() as db: service = AgentAssetService(db) rule = next( item for item in service.list_assets(asset_type=AgentAssetType.RULE.value) if item.code == "rule.expense.company_travel_expense_reimbursement" ) service.upload_rule_spreadsheet( rule.id, filename="公司差旅费报销规则.xlsx", content=build_workbook_bytes([["城市", "住宿"], ["北京", 500]]), actor="finance_user", ) detail = service.get_asset(rule.id) assert detail is not None current_asset = service.repository.get(rule.id) assert current_asset is not None live_storage_key = str((current_asset.config_json or {})["rule_document"]["storage_key"]) assert live_storage_key.startswith(f"rules/{FINANCE_RULES_LIBRARY}/") assert "agent_assets" not in live_storage_key live_path = service.spreadsheet_manager.resolve_storage_path(live_storage_key) assert not service.spreadsheet_manager.asset_root.exists() current_path, _, _ = service.get_rule_spreadsheet_content(rule.id) assert current_path == live_path assert ".versions" not in current_path.parts workbook = load_workbook(current_path, data_only=False) assert workbook.active["B2"].value == 500 def test_spreadsheet_change_records_return_recent_edit_details() -> None: with build_session() as db: service = AgentAssetService(db) rule = next( item for item in service.list_assets(asset_type=AgentAssetType.RULE.value) if item.code == "rule.expense.company_travel_expense_reimbursement" ) service.audit_service.log_action( actor="manager_user", action="edit_rule_spreadsheet", resource_type=rule.asset_type, resource_id=rule.id, after_json={ "summary": "在线编辑：共 1 处改动。", "changed_sheet_count": 1, "changed_cell_count": 1, "sheet_changes": [], "cell_changes": [ { "sheet_name": "规则表", "cell": "B2", "change_type": "modified", "before_value": 500, "after_value": 550, } ], }, ) records = service.list_spreadsheet_change_records(rule.id) assert len(records) == 1 assert records[0].actor == "manager_user" assert records[0].changed_cell_count == 1 assert records[0].cell_changes[0].cell == "B2" def test_spreadsheet_change_records_include_all_modified_sheets() -> None: with build_session() as db: service = AgentAssetService(db) rule = next( item for item in service.list_assets(asset_type=AgentAssetType.RULE.value) if item.code == "rule.expense.company_travel_expense_reimbursement" ) service.upload_rule_spreadsheet( rule.id, filename="公司差旅费报销规则.xlsx", content=build_multi_sheet_workbook_bytes( { "差旅标准": [["城市", "住宿"], ["北京", 500]], "填表说明": [["字段", "说明"], ["住宿", "按城市标准"]], } ), actor="finance_user", ) detail = service.get_asset(rule.id) assert detail is not None service.upload_rule_spreadsheet( rule.id, filename="公司差旅费报销规则.xlsx", content=build_multi_sheet_workbook_bytes( { "差旅标准": [["城市", "住宿"], ["北京", 550]], "填表说明": [["字段", "说明"], ["住宿", "按城市等级标准"]], } ), actor="finance_user", ) records = service.list_spreadsheet_change_records(rule.id) latest = records[0] changed_sheets = {item.sheet_name for item in latest.sheet_changes} changed_cell_sheets = {item.sheet_name for item in latest.cell_changes} assert not hasattr(latest, "version") assert latest.changed_sheet_count == 2 assert {"差旅标准", "填表说明"}.issubset(changed_sheets) assert {"差旅标准", "填表说明"}.issubset(changed_cell_sheets) assert "差旅标准" in latest.summary assert "填表说明" in latest.summary def test_editable_spreadsheet_onlyoffice_config_enables_forcesave(monkeypatch) -> None: with build_session() as db: monkeypatch.setattr( "app.services.agent_asset_onlyoffice.resolve_onlyoffice_settings", lambda: OnlyOfficeRuntimeConfig( enabled=True, public_url="http://onlyoffice.example.com", backend_url="http://backend.example.com", jwt_secret="secret", ), ) service = AgentAssetService(db) rule = next( item for item in service.list_assets(asset_type=AgentAssetType.RULE.value) if item.code == "rule.expense.company_travel_expense_reimbursement" ) config = service.build_rule_spreadsheet_onlyoffice_config( rule.id, CurrentUserContext( username="finance_user", name="财务人员", role_codes=["finance"], is_admin=False, ), ) customization = config.config["editorConfig"]["customization"] assert config.config["editorConfig"]["mode"] == "edit" assert customization["forcesave"] is True assert "version=" not in config.config["document"]["url"] assert "version=" not in config.config["editorConfig"]["callbackUrl"] def test_version_timeline_contains_created_review_and_publish_events() -> None: with build_session() as db: service = AgentAssetService(db) rule = next( item for item in service.list_assets(asset_type=AgentAssetType.RULE.value) if item.code == "rule.expense.travel_risk_control_standard" ) service.create_version( rule.id, AgentAssetVersionCreate( version="v1.1.1", content="# 工作稿", content_type=AgentAssetContentType.MARKDOWN, change_note="补充口径", created_by="finance_user", ), actor="finance_user", ) service.create_review( rule.id, AgentAssetReviewCreate( version="v1.1.1", reviewer="finance_user", review_status=AgentReviewStatus.PENDING, review_note="请审核", ), actor="finance_user", ) service.create_review( rule.id, AgentAssetReviewCreate( version="v1.1.1", reviewer="manager_user", review_status=AgentReviewStatus.APPROVED, review_note="可以上线", ), actor="manager_user", ) service.activate_asset(rule.id, actor="manager_user") timeline = service.list_version_timeline(rule.id) event_types = [item.event_type for item in timeline if item.version == "v1.1.1"] assert "created" in event_types assert "submitted" in event_types assert "approved" in event_types assert "published" in event_types def test_agent_asset_service_returns_recent_versions_for_rule_detail() -> None: with build_session() as db: service = AgentAssetService(db) rule = next( item for item in service.list_assets(asset_type=AgentAssetType.RULE.value) if item.code == "rule.expense.attachment_submission_requirements" ) detail = service.get_asset(rule.id) assert detail is not None assert detail.current_version == "v1.0.0" assert detail.current_version_content_type == AgentAssetContentType.MARKDOWN.value assert isinstance(detail.current_version_content, str) assert len(detail.recent_versions) >= 2 assert any(item.is_current for item in detail.recent_versions) assert {item.version for item in detail.recent_versions} >= {"v0.9.0", "v1.0.0"} assert detail.config_json["rule_template_key"] == "attachment_requirement_v1" assert "附件或单据不完整" in str(detail.current_version_content) def test_agent_asset_service_returns_travel_policy_rule_detail() -> None: with build_session() as db: service = AgentAssetService(db) rule = next( item for item in service.list_assets(asset_type=AgentAssetType.RULE.value) if item.code == "rule.expense.travel_risk_control_standard" ) detail = service.get_asset(rule.id) assert detail is not None assert detail.status == AgentAssetStatus.ACTIVE.value assert detail.current_version == "v1.1.0" assert detail.latest_review is not None assert detail.latest_review.review_status == AgentReviewStatus.APPROVED.value assert "行程闭环" in str(detail.current_version_content) assert "住宿标准、飞机舱位和火车席别" in str(detail.current_version_content) def test_expense_rule_runtime_reads_amount_standards_from_travel_spreadsheet() -> None: with build_session() as db: AgentAssetService(db).list_assets(asset_type=AgentAssetType.RULE.value) travel_spreadsheet_rule = db.scalar( select(AgentAsset).where(AgentAsset.code == COMPANY_TRAVEL_EXPENSE_RULE_CODE) ) assert travel_spreadsheet_rule is not None travel_spreadsheet_rule.status = AgentAssetStatus.REVIEW.value db.commit() catalog = ExpenseRuleRuntimeService(db).load_catalog() assert catalog.travel_policy is not None assert catalog.travel_policy.standard_rule_code == COMPANY_TRAVEL_EXPENSE_RULE_CODE assert catalog.travel_policy.standard_rule_name == "公司差旅费报销规则" assert catalog.travel_policy.hotel_city_limits["北京"]["mid"] == 450 assert catalog.travel_policy.hotel_city_limits["北京"]["junior"] == 450 assert catalog.travel_policy.hotel_city_limits["北京"]["manager"] == 500 assert catalog.travel_policy.allowance_limits["meal"]["直辖市/特区"] == 65 assert catalog.travel_policy.allowance_limits["meal"]["其他地区"] == 55 assert catalog.travel_policy.allowance_limits["total"]["其他地区"] == 90 assert catalog.travel_policy.transport_limits["senior"]["flight"] == 1 assert catalog.travel_policy.transport_limits["executive"]["train"] == 1 def test_travel_reimbursement_calculator_uses_finance_spreadsheet_amounts() -> None: with build_session() as db: db.add( Employee( employee_no="E9001", name="测试员工", email="traveler@example.com", position="产品经理", grade="P4", ) ) db.commit() result = TravelReimbursementCalculatorService(db).calculate( TravelReimbursementCalculatorRequest(days=3, location="北京市朝阳区"), CurrentUserContext( username="traveler@example.com", name="测试员工", role_codes=[], is_admin=False, ), ) assert result.rule_name == "公司差旅费报销规则" assert result.grade == "P4" assert result.grade_band == "mid" assert result.matched_city == "北京" assert result.hotel_rate == 450 assert result.hotel_amount == 1350 assert result.allowance_region == "直辖市/特区" assert result.total_allowance_rate == 100 assert result.allowance_amount == 300 assert result.total_amount == 1650 assert "住宿 450.00 × 3 天 + 补贴 100.00 × 3 天 = 1650.00" == result.formula_text assert "参考可报销总金额为 1650.00 元" in result.summary_text def test_travel_reimbursement_calculator_uses_other_region_for_known_unlisted_location() -> None: with build_session() as db: db.add( Employee( employee_no="E9002", name="其他地区员工", email="other-region@example.com", position="产品经理", grade="P4", ) ) db.commit() result = TravelReimbursementCalculatorService(db).calculate( TravelReimbursementCalculatorRequest(days=2, location="吉林延边"), CurrentUserContext( username="other-region@example.com", name="其他地区员工", role_codes=[], is_admin=False, ), ) assert result.matched_city == "延边（其他地区）" assert result.city_tier == "tier_3" assert result.hotel_rate == 380 assert result.hotel_amount == 760 assert result.allowance_region == "其他地区" assert result.total_allowance_rate == 90 assert result.allowance_amount == 180 assert result.total_amount == 940 def test_travel_reimbursement_calculator_rejects_unrecognized_location() -> None: with build_session() as db: db.add( Employee( employee_no="E9003", name="无效地点员工", email="invalid-location@example.com", position="产品经理", grade="P4", ) ) db.commit() with pytest.raises(ValueError, match="未识别为有效出差地区"): TravelReimbursementCalculatorService(db).calculate( TravelReimbursementCalculatorRequest(days=2, location="背景"), CurrentUserContext( username="invalid-location@example.com", name="无效地点员工", role_codes=[], is_admin=False, ), ) def test_agent_run_service_lists_seeded_trace_data() -> None: with build_session() as db: service = AgentRunService(db) runs = service.list_runs() assert len(runs) >= 3 assert any(item.tool_calls for item in runs) assert any(item.semantic_parse is not None for item in runs) def test_agent_run_service_creates_run_and_persists_error_message() -> None: with build_session() as db: service = AgentRunService(db) created = service.create_run( agent=AgentName.ORCHESTRATOR.value, source=AgentRunSource.SYSTEM_EVENT.value, status=AgentRunStatus.FAILED.value, error_message="simulated failure", result_summary="failed to route request", ) fetched = service.get_run(created.run_id) assert fetched is not None assert fetched.run_id.startswith("run_") assert fetched.status == AgentRunStatus.FAILED.value assert fetched.error_message == "simulated failure" assert fetched.result_summary == "failed to route request" def test_agent_asset_creation_writes_audit_log() -> None: with build_session() as db: service = AgentAssetService(db) created = service.create_asset( AgentAssetCreate( asset_type=AgentAssetType.SKILL, code=f"skill.test.{uuid.uuid4().hex[:8]}", name="测试技能", description="用于测试审计日志写入。", domain=AgentAssetDomain.KNOWLEDGE, scenario_json=["knowledge", "query"], owner="pytest", reviewer="reviewer", status=AgentAssetStatus.DRAFT, config_json={"enabled": True}, ), actor="pytest", ) logs = AuditLogService(db).list_logs(resource_id=created.id) assert any(item.action == "create_agent_asset" for item in logs)