from __future__ import annotations import json from pathlib import Path from types import SimpleNamespace import pytest from fastapi import FastAPI from httpx import ASGITransport, AsyncClient from app.api.ai_steward_api import routes_capture from app.api.ai_steward_api.schemas import ProjectFromTextResponse from app.core.database import get_db from app.dependencies.auth import get_current_user from app.dependencies.tenant import get_tenant_id from app.services.ai_steward_service import mixin_creation_normalize_persist from app.services.ai_steward_service import AIStewardService from app.services.task_persistence_core import SkillTagResolution @pytest.mark.asyncio async def test_project_from_text_returns_enveloped_signal_sources(monkeypatch) -> None: async def fake_create_project_from_natural_language(self, **kwargs): # noqa: ANN001 _ = (self, kwargs) return { "project": {"id": "11", "name": "AI 客服", "title": "AI 客服平台", "description": "demo", "status": "active"}, "signal": {"id": "21", "content": "main signal", "impact_score": 0.91}, "charter": {"id": "31", "title": "宪章", "content": {"summary": "demo"}}, "tasks": [], "risks": [], "analysis": { "project_name": "AI 客服", "project_title": "AI 客服平台", "tech_stack": ["FastAPI", "React"], "timeline": "3 周", "primary_role": "backend", "requirement_items": [ { "section": "核心痛点", "title": "内容本地化", "description": "翻译公司贵、慢、不地道", } ], }, "signal_sources": [ { "id": "101", "source_kind": "custom", "source_name": "核心痛点 - 内容本地化", "source_identifier": "custom:core-pain-content-localization", "display_source_kind": "自定义来源", "technical_identifier": "custom:core-pain-content-localization", "description": "翻译公司贵、慢、不地道", "section": "核心痛点", } ], } async def _fake_get_db(): yield object() monkeypatch.setattr(AIStewardService, "create_project_from_natural_language", fake_create_project_from_natural_language) app = FastAPI() app.include_router(routes_capture.router) app.dependency_overrides[get_db] = _fake_get_db app.dependency_overrides[get_tenant_id] = lambda: 1 app.dependency_overrides[get_current_user] = lambda: SimpleNamespace( id=9, tenant_id=1, name="管理员", username="admin", role="ADMIN", ) transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: response = await client.post("/ai-steward/project-from-text", json={"input_text": "我们要做一个 AI 客服工具"}) assert response.status_code == 200 payload = response.json() assert payload["code"] == 0 assert payload["message"] == "success" assert payload["data"]["project"]["id"] == "11" assert payload["data"]["signal_sources"][0]["source_name"] == "核心痛点 - 内容本地化" assert payload["data"]["signal_sources"][0]["display_source_kind"] == "自定义来源" assert payload["data"]["signal_sources"][0]["technical_identifier"] == "custom:core-pain-content-localization" def test_project_from_text_response_schema_exposes_signal_source_display_fields() -> None: schema = ProjectFromTextResponse.model_json_schema() signal_source = schema["$defs"]["ProjectFromTextSignalSource"] assert "display_source_kind" in signal_source["properties"] assert "technical_identifier" in signal_source["properties"] def test_ai_steward_no_longer_keeps_instruction_or_stub_heuristics_in_runtime_modules() -> None: repo_root = Path(__file__).resolve().parents[1] title_utils = (repo_root / "app/services/ai_steward_service/mixin_title_utils.py").read_text(encoding="utf-8") json_parsing = (repo_root / "app/services/ai_steward_service/mixin_json_parsing.py").read_text(encoding="utf-8") shared = (repo_root / "app/services/ai_steward_service/shared.py").read_text(encoding="utf-8") assert "_looks_like_instruction_sentence" not in title_utils assert "_looks_like_project_creation_stub_response" not in json_parsing assert "PROJECT_CREATION_STUB_MARKERS" not in shared @pytest.mark.asyncio async def test_project_requirement_analysis_uses_canonical_function_name(monkeypatch) -> None: service = AIStewardService(object()) captured: dict[str, object] = {} async def fake_generate(prompt: str, **kwargs): # noqa: ANN001 _ = prompt captured["function_name"] = kwargs.get("function_name") captured["max_tokens"] = kwargs.get("max_tokens") return '{"project_title":"AI 客服","requirement_items":[]}' monkeypatch.setattr(service.llm, "generate", fake_generate) analysis = await service._analyze_project_input("请创建一个 AI 客服项目", tenant_id=1) assert analysis["project_title"] == "AI 客服" assert captured["function_name"] == "project_requirement_extraction" assert captured["max_tokens"] == 2_200 @pytest.mark.asyncio async def test_project_requirement_analysis_escalates_token_budget_low_to_high(monkeypatch) -> None: service = AIStewardService(object()) token_attempts: list[int] = [] async def fake_generate(prompt: str, **kwargs): # noqa: ANN001 _ = prompt token_attempts.append(int(kwargs.get("max_tokens") or 0)) if len(token_attempts) == 1: return '{"project_title":"AI 客服"' return '{"project_title":"AI 客服","requirement_items":[]}' monkeypatch.setattr(service.llm, "generate", fake_generate) analysis = await service._analyze_project_input("请创建一个 AI 客服项目", tenant_id=1) assert analysis["project_title"] == "AI 客服" assert token_attempts[:2] == [2_200, 4_000] @pytest.mark.asyncio async def test_normalize_project_creation_draft_auto_assigns_best_skill_match(monkeypatch) -> None: service = AIStewardService(object()) captured: dict[str, object] = {} async def fake_resolve_skill_tags(db, *, tenant_id: int, skill_names): # noqa: ANN001 _ = db captured["resolved_tenant_id"] = tenant_id captured["skill_names"] = list(skill_names) return SkillTagResolution( input_names=list(skill_names), matched_ids=[101, 202], matched_items=[ {"id": 101, "name": "FastAPI"}, {"id": 202, "name": "React"}, ], unmatched_names=[], ) async def fake_recommend_personnel( db, # noqa: ANN001 tenant_id: int, required_skills, task_description=None, task_id=None, limit: int = 5, ): _ = (db, task_description, task_id, limit) captured["recommend_tenant_id"] = tenant_id captured["required_skills"] = list(required_skills) return { "task_id": None, "task_description": "搭建 AI 客服核心接口", "required_skills": list(required_skills), "recommendations": [ { "user_id": "9", "user_name": "Ada", "match_score": 95.0, "reason": "FastAPI 和 React 都命中，且负载适中", "skill_match_count": 2, "avg_proficiency": 88.0, "workload_rate": 45.0, "recommended_role": "backend", } ], "total_count": 1, } monkeypatch.setattr( mixin_creation_normalize_persist, "resolve_skill_tag_ids_by_names", fake_resolve_skill_tags, ) monkeypatch.setattr( mixin_creation_normalize_persist, "recommend_personnel", fake_recommend_personnel, ) normalized = await service._normalize_project_creation_draft( draft={ "analysis": {}, "project": { "name": "AI 客服", "title": "AI 客服平台", "description": "创建一个 AI 客服平台", "assigned_role": "backend", }, "signal": { "content": "main signal", "impact_score": 0.9, "priority": "high", "strategic_analysis": {}, }, "charter": { "title": "AI 客服平台宪章", "content": { "objectives": ["提升客服效率"], "success_criteria": ["首响 < 10 秒"], "constraints": ["先做 MVP"], }, }, "tasks": [ { "title": "搭建 AI 客服核心接口", "description": "实现工单接入与回复编排", "priority": "high", "required_skills": ["FastAPI", "React"], } ], "risks": [], }, input_text="创建一个 AI 客服平台", creator_id=1, tenant_id=1, ) task_draft = normalized["tasks"][0] assert task_draft.assignee_id == 9 assert task_draft.required_skill_tag_ids == [101, 202] assert task_draft.ai_recommended_assignees is not None assert task_draft.ai_recommended_assignees["final_assignee_id"] == 9 assert task_draft.ai_recommended_assignees["auto_selected"] is True assert task_draft.ai_recommended_assignees["recommended_candidates"][0]["user_id"] == "9" assert captured["skill_names"] == ["FastAPI", "React"] assert captured["required_skills"] == [101, 202] @pytest.mark.asyncio async def test_normalize_project_creation_draft_prefers_explicit_fixed_project_title() -> None: service = AIStewardService(object()) normalized = await service._normalize_project_creation_draft( draft={ "analysis": {}, "project": { "name": "跨境AI复盘", "title": "跨境AI复盘", "description": "通用跨境直播复盘项目", "assigned_role": "backend", }, "signal": { "content": "main signal", "impact_score": 0.9, "priority": "high", "strategic_analysis": {}, }, "charter": { "title": "跨境AI复盘项目宪章", "content": { "objectives": ["提升复盘效率"], "success_criteria": ["复盘耗时下降"], "constraints": ["先做 MVP"], }, }, "tasks": [], "risks": [], }, input_text=( "请发起一个全新的项目需求，项目名固定为 FeishuWebFlow-1775111724952。 " "这是一个从 0 到 1 的需求：做面向跨境 DTC 团队的 AI 直播复盘助手。" ), creator_id=1, tenant_id=1, ) assert normalized["project"]["title"] == "FeishuWebFlow-1775111724952" assert normalized["project"]["name"] == "FeishuWebFlow-1775111724952" assert normalized["charter"]["title"] == "FeishuWebFlow-1775111724952项目宪章" @pytest.mark.asyncio async def test_normalize_project_creation_draft_prefers_natural_named_project_title() -> None: service = AIStewardService(object()) normalized = await service._normalize_project_creation_draft( draft={ "analysis": {}, "project": { "name": "跨境复盘快版", "title": "跨境复盘快版", "description": "为北美 DTC 运营团队提供 3 分钟级 AI 复盘工具", "assigned_role": "backend", }, "signal": { "content": "main signal", "impact_score": 0.9, "priority": "high", "strategic_analysis": {}, }, "charter": { "title": "跨境复盘快版项目宪章", "content": { "objectives": ["提升复盘效率"], "success_criteria": ["复盘耗时下降"], "constraints": ["先做 MVP"], }, }, "tasks": [], "risks": [], }, input_text=( "最近我们做跨境直播复盘太慢了，想做个给运营团队用的小工具，" "这个需求先叫 FeishuWebFlow-1775134530933。" ), creator_id=1, tenant_id=1, ) assert normalized["project"]["title"] == "FeishuWebFlow-1775134530933" assert normalized["project"]["name"] == "FeishuWebFlow-1775134530933" assert normalized["charter"]["title"] == "FeishuWebFlow-1775134530933项目宪章" @pytest.mark.asyncio async def test_generate_task_chain_prompt_includes_rich_requirement_context_and_skill_catalog( monkeypatch, ) -> None: service = AIStewardService(object()) captured: dict[str, object] = {} async def fake_load_available_skill_names(self, *, tenant_id: int, limit: int = 20): # noqa: ANN001 _ = self, limit captured["skill_catalog_tenant_id"] = tenant_id return ["Python", "FastAPI", "React", "TypeScript", "LLM 调优"] async def fake_generate(prompt: str, **kwargs): # noqa: ANN001 captured["prompt"] = prompt captured["function_name"] = kwargs.get("function_name") return json.dumps( { "tasks": [ { "title": "直播数据采集与清洗", "type": "backend", "estimated_hours": 16, "description": "接入录屏、评论与交易数据并完成清洗。", "priority": "high", "required_skills": ["Python", "FastAPI"], } ] }, ensure_ascii=False, ) async def fake_resolve_skill_tags(db, *, tenant_id: int, skill_names): # noqa: ANN001 _ = db, tenant_id input_names = list(skill_names) return SkillTagResolution( input_names=input_names, matched_ids=[1, 3], matched_items=[ {"input_name": "Python", "skill_tag_id": 1, "skill_tag_name": "Python", "skill_tag_slug": "python"}, {"input_name": "FastAPI", "skill_tag_id": 3, "skill_tag_name": "FastAPI", "skill_tag_slug": "fastapi"}, ], unmatched_names=[], ) monkeypatch.setattr(AIStewardService, "_load_available_skill_names", fake_load_available_skill_names) monkeypatch.setattr( "app.services.ai_steward_service.mixin_draft_generation.resolve_skill_tag_ids_by_names", fake_resolve_skill_tags, ) monkeypatch.setattr(service.llm, "generate", fake_generate) tasks = await service._generate_task_chain_draft( analysis={ "project_title": "跨境直播复盘", "timeline": "8周", "target_audience": "北美 DTC 品牌运营团队", "key_features": [ "直播录屏与评论回放切片", "亮点和掉点自动总结", "可执行复盘清单", ], "success_metrics": [ "7天内复盘时长下降60%", "高频问题复现率下降30%", ], "team_setup": [ {"role": "PM", "notes": "负责需求边界与验收口径"}, {"role": "后端", "notes": "负责直播数据与订单数据接入"}, ], "requirement_items": [ { "section": "信号源", "title": "Shopify 订单与退款数据", "description": "用于转化归因与退款波动分析", }, { "section": "信号源", "title": "TikTok Shop 直播数据", "description": "提供观看、互动与成交数据", }, ], }, tenant_id=1, ) assert tasks[0]["title"] == "直播数据采集与清洗" assert captured["function_name"] == "project_from_text_task_chain" assert captured["skill_catalog_tenant_id"] == 1 prompt = str(captured["prompt"] or "") assert "直播录屏与评论回放切片" in prompt assert "7天内复盘时长下降60%" in prompt assert "Shopify 订单与退款数据" in prompt assert "PM" in prompt assert "Python" in prompt assert "FastAPI" in prompt assert "LLM 调优" in prompt @pytest.mark.asyncio async def test_generate_task_chain_retries_when_first_result_has_only_unmappable_skills(monkeypatch) -> None: service = AIStewardService(object()) prompts: list[str] = [] async def fake_load_available_skill_names(self, *, tenant_id: int, limit: int = 20): # noqa: ANN001 _ = self, tenant_id, limit return ["Python", "FastAPI", "React", "TypeScript", "LLM 调优"] async def fake_resolve_skill_tags(db, *, tenant_id: int, skill_names): # noqa: ANN001 _ = db, tenant_id input_names = list(skill_names) mapping = { "Python": (1, "Python"), "FastAPI": (3, "FastAPI"), "React": (4, "React"), "TypeScript": (2, "TypeScript"), "LLM 调优": (7, "LLM 调优"), } matched_items = [] matched_ids = [] unmatched_names = [] for item in input_names: matched = mapping.get(str(item)) if matched is None: unmatched_names.append(str(item)) continue matched_ids.append(matched[0]) matched_items.append( { "input_name": str(item), "skill_tag_id": matched[0], "skill_tag_name": matched[1], "skill_tag_slug": matched[1].lower(), } ) return SkillTagResolution( input_names=input_names, matched_ids=matched_ids, matched_items=matched_items, unmatched_names=unmatched_names, ) invalid_response = json.dumps( { "tasks": [ { "title": "用户权限管理模块", "type": "backend", "estimated_hours": 20, "description": "实现用户登录鉴权与角色权限管理。", "priority": "low", "required_skills": ["JWT", "Node.js", "Database Design"], } ] }, ensure_ascii=False, ) valid_response = json.dumps( { "tasks": [ { "title": "直播数据采集与清洗", "type": "backend", "estimated_hours": 24, "description": "接入直播录屏与评论数据并完成清洗。", "priority": "high", "required_skills": ["Python", "FastAPI"], }, { "title": "复盘报告看板", "type": "frontend", "estimated_hours": 24, "description": "展示亮点、掉点与复盘清单。", "priority": "medium", "required_skills": ["React", "TypeScript"], }, ] }, ensure_ascii=False, ) async def fake_generate(prompt: str, **kwargs): # noqa: ANN001 _ = kwargs prompts.append(prompt) return invalid_response if len(prompts) == 1 else valid_response monkeypatch.setattr(AIStewardService, "_load_available_skill_names", fake_load_available_skill_names) monkeypatch.setattr( "app.services.ai_steward_service.mixin_draft_generation.resolve_skill_tag_ids_by_names", fake_resolve_skill_tags, ) monkeypatch.setattr(service.llm, "generate", fake_generate) tasks = await service._generate_task_chain_draft( analysis={ "project_title": "跨境直播复盘", "timeline": "8周", "key_features": [ "直播录屏与评论回放切片", "亮点和掉点自动总结", ], "success_metrics": ["7天内复盘时长下降60%"], "requirement_items": [ { "section": "MVP 功能", "title": "可执行复盘清单", "description": "输出待办清单与复盘建议", } ], }, tenant_id=1, ) assert len(prompts) == 2 assert tasks[0]["title"] == "直播数据采集与清洗" assert tasks[0]["required_skills"] == ["Python", "FastAPI"] assert "用户权限管理模块" in prompts[1] assert "JWT" in prompts[1] assert "Python" in prompts[1] assert "FastAPI" in prompts[1] @pytest.mark.asyncio async def test_generate_task_chain_falls_back_to_charter_decomposition_when_primary_route_fails( monkeypatch, ) -> None: service = AIStewardService(object()) captured: dict[str, object] = {} async def fake_load_available_skill_names(self, *, tenant_id: int, limit: int = 20): # noqa: ANN001 _ = self, tenant_id, limit return ["Python", "FastAPI", "React", "TypeScript", "LLM 调优"] async def fake_generate_project_creation_object(self, **kwargs): # noqa: ANN001 _ = self captured["primary_function_name"] = kwargs.get("function_name") raise RuntimeError("provider_circuit_open:openai") async def fake_decompose_tasks_from_charter(self, charter_content, project_id, tenant_id="default", db=None): # noqa: ANN001 _ = self, project_id, db captured["charter_content"] = charter_content captured["decompose_tenant_id"] = tenant_id return { "tasks": [ { "task_key": "task-1", "title": "直播数据采集与清洗", "description": "接入直播录屏与评论数据并完成清洗。", "type": "backend", "estimated_hours": 24, "required_skill_names": ["Python", "FastAPI"], "priority": "high", }, { "task_key": "task-2", "title": "复盘报告看板", "description": "展示亮点、掉点与复盘清单。", "type": "frontend", "estimated_hours": 20, "required_skill_names": ["React", "TypeScript"], "priority": "medium", }, ] } monkeypatch.setattr(AIStewardService, "_load_available_skill_names", fake_load_available_skill_names) monkeypatch.setattr(AIStewardService, "_generate_project_creation_object", fake_generate_project_creation_object) monkeypatch.setattr( "app.services.ai_service_core.service.AIService.decompose_tasks_from_charter", fake_decompose_tasks_from_charter, ) tasks = await service._generate_task_chain_draft( analysis={ "project_title": "跨境直播复盘", "target_audience": "北美 DTC 品牌运营团队", "key_features": ["直播录屏与评论回放切片", "亮点和掉点自动总结", "可执行复盘清单"], "success_metrics": ["7天内复盘时长下降60%", "高频问题复现率下降30%"], "requirement_items": [ { "section": "信号源", "title": "Shopify 订单与退款数据", "description": "用于转化归因与退款波动分析", } ], "team_setup": [{"role": "后端", "notes": "负责直播数据与订单数据接入"}], }, tenant_id=1, ) assert captured["primary_function_name"] == "project_from_text_task_chain" assert captured["decompose_tenant_id"] == 1 assert "直播录屏与评论回放切片" in str(captured["charter_content"] or "") assert tasks[0]["title"] == "直播数据采集与清洗" assert tasks[0]["required_skills"] == ["Python", "FastAPI"] @pytest.mark.asyncio async def test_generate_task_chain_prefers_six_module_charter_content_when_present(monkeypatch) -> None: service = AIStewardService(object()) captured: dict[str, object] = {} async def fake_load_available_skill_names(self, *, tenant_id: int, limit: int = 24): # noqa: ANN001 _ = self, tenant_id, limit return ["Python", "FastAPI", "React", "TypeScript"] async def fake_generate_project_creation_object(self, **kwargs): # noqa: ANN001 _ = self, kwargs raise AssertionError("six-module charter path should bypass project_from_text_task_chain prompt") async def fake_decompose_tasks_from_charter(self, charter_content, project_id, tenant_id="default", db=None): # noqa: ANN001 _ = self, project_id, db captured["charter_content"] = charter_content captured["decompose_tenant_id"] = tenant_id return { "tasks": [ { "task_key": "task-1", "title": "直播数据采集与清洗", "description": "接入直播录屏与评论数据并完成清洗。", "type": "backend", "estimated_hours": 24, "required_skill_names": ["Python", "FastAPI"], "priority": "high", } ] } monkeypatch.setattr(AIStewardService, "_load_available_skill_names", fake_load_available_skill_names) monkeypatch.setattr(AIStewardService, "_generate_project_creation_object", fake_generate_project_creation_object) monkeypatch.setattr( "app.services.ai_service_core.service.AIService.decompose_tasks_from_charter", fake_decompose_tasks_from_charter, ) tasks = await service._generate_task_chain_draft( analysis={ "project_title": "跨境直播复盘", "target_audience": "北美 DTC 品牌运营团队", "key_features": ["直播录屏与评论回放切片", "亮点和掉点自动总结", "可执行复盘清单"], }, tenant_id=1, charter_content={ "schema_version": "charter_ai_first_v1", "modules": { "module_1": { "problem_statement": "北美 DTC 品牌运营团队直播结束后复盘慢、素材复用率低。", "goal_statement": "- 将复盘压缩到 10 分钟内", "kpis": "- 7 天内复盘时长下降 60%", }, "module_2": { "in_scope": "- 直播录屏与评论回放切片\n- 亮点和掉点自动总结", "out_of_scope": "- 不做达人 CRM", "alignment_qa": "Q1: why now", }, "module_3": {}, "module_4": {"milestones": "- 第 1 周完成数据接入"}, "module_5": {"risk_register": "- 数据口径不一致"}, "module_6": {"governance_rule": "- 复盘记录保留 180 天"}, }, }, ) assert captured["decompose_tenant_id"] == 1 assert "\"modules\"" in str(captured["charter_content"] or "") assert "直播录屏与评论回放切片" in str(captured["charter_content"] or "") assert tasks[0]["title"] == "直播数据采集与清洗" assert tasks[0]["required_skills"] == ["Python", "FastAPI"] @pytest.mark.asyncio async def test_generate_task_chain_raises_when_charter_decomposition_provider_fails(monkeypatch) -> None: service = AIStewardService(object()) async def fake_decompose_tasks_from_charter(self, charter_content, project_id, tenant_id="default", db=None): # noqa: ANN001 _ = self, charter_content, project_id, tenant_id, db raise RuntimeError("provider_circuit_open:openai") monkeypatch.setattr( "app.services.ai_service_core.service.AIService.decompose_tasks_from_charter", fake_decompose_tasks_from_charter, ) with pytest.raises(RuntimeError, match="provider_circuit_open:openai"): await service._generate_task_chain_draft( analysis={ "project_title": "跨境直播复盘", "primary_role": "backend", }, tenant_id=1, charter_content={ "schema_version": "charter_ai_first_v1", "modules": { "module_1": { "problem_statement": "直播复盘慢，问题复现率高。", "goal_statement": "- 将复盘压缩到 10 分钟内", "kpis": "- 7 天内复盘时长下降 60%", }, "module_2": { "in_scope": "- 直播录屏与评论回放切片\n- 亮点和掉点自动总结\n- 可执行复盘清单", "out_of_scope": "- 不做达人 CRM", "alignment_qa": "优先交付复盘效率闭环", }, "module_3": { "engineering_view": "Python FastAPI React TypeScript 数据分析", }, "module_4": { "milestones": "- 第 1 周完成直播与订单数据对齐\n- 第 2 周完成复盘看板 MVP", }, "module_5": { "risk_register": "- 数据口径不一致", }, "module_6": { "governance_rule": "- 复盘记录保留 180 天", }, }, "derived_artifacts": { "task_seeds": [ { "title": "直播录屏与评论回放切片", "description": "接入录屏与评论数据，支持按时间轴回放。", "evidence_refs": ["module_2.in_scope"], }, { "title": "亮点和掉点自动总结", "description": "自动总结直播亮点、掉点与问题复现线索。", "evidence_refs": ["module_1.goal_statement", "module_5.risk_register"], }, ] }, }, )