import json from types import SimpleNamespace import pytest from app.api import tasks as tasks_api from app.api.sstc_api import tasks_ai_breakdown as sstc_tasks_ai_breakdown_api from app.api.sstc_api import tasks_recommend_optimize as sstc_tasks_recommend_optimize_api from app.services.llm_service import StructuredExtractionResult from app.services.sstc.charter_data import ( _build_task_breakdown_context, _build_task_breakdown_prompt, ) from app.schemas import TaskBreakdownRequest def test_task_breakdown_context_flattens_charter_modules() -> None: project = SimpleNamespace( id=6, title="内部知识库升级", name="内部知识库升级", description="围绕内部知识库问答能力升级", status="active", assigned_role="backend", ) charter = SimpleNamespace( id=55, title="内部知识库升级宪章", version=2, status="aligning", content={ "modules": { "module_1": { "problem_statement": "减少重复检索", "goal_statement": "提升命中率 / 降低重复提问", "kpis": "首响 < 3s", }, "module_2": { "in_scope": "问答检索，知识补充", "out_of_scope": "权限模型重构", }, "module_4": { "milestones": "需求冻结\n开发完成", }, "module_5": { "risk_register": "需求膨胀; 依赖不清", }, "module_6": { "governance_rule": "附件仅作证据引用", }, } }, ) context = _build_task_breakdown_context( description="支持导出知识库内容", project=project, charter=charter, ) assert context["project"]["title"] == "内部知识库升级" assert context["project"]["status"] == "active" assert context["charter"]["summary"] == "减少重复检索" assert context["charter"]["objectives"] == ["提升命中率", "降低重复提问"] assert context["charter"]["in_scope"] == ["问答检索", "知识补充"] assert context["charter"]["out_of_scope"] == ["权限模型重构"] assert context["charter"]["milestones"] == ["需求冻结", "开发完成"] assert "附件仅作证据引用" in context["charter"]["constraints"][0] def test_task_breakdown_context_prefers_derived_task_seeds() -> None: project = SimpleNamespace( id=6, title="内部知识库升级", name="内部知识库升级", description="围绕内部知识库问答能力升级", status="active", assigned_role="backend", ) charter = SimpleNamespace( id=55, title="内部知识库升级宪章", version=2, status="aligned", content={ "modules": { "module_1": { "problem_statement": "减少重复检索", "goal_statement": "提升命中率 / 降低重复提问", "kpis": "首响 < 3s", }, "module_4": { "milestones": "需求冻结\n开发完成", }, }, "derived_artifacts": { "task_seeds": [ { "title": "先补检索评估基线", "description": "补齐离线评测与验收口径", "evidence_refs": ["module_1.kpis"], }, { "title": "再实现导出链路", "description": "基于基线结果实现最小闭环", "evidence_refs": ["module_4.milestones"], }, ] }, }, ) context = _build_task_breakdown_context( description="支持导出知识库内容", project=project, charter=charter, ) assert context["charter"]["task_seeds"][0]["title"] == "先补检索评估基线" assert context["charter"]["task_seeds"][1]["evidence_refs"] == ["module_4.milestones"] def test_task_breakdown_prompt_embeds_charter_context() -> None: prompt = _build_task_breakdown_prompt( project_id=6, context={ "project": { "id": 6, "title": "内部知识库升级", "description": "围绕内部知识库问答能力升级", "status": "active", "assigned_role": "backend", }, "charter": { "id": 55, "title": "内部知识库升级宪章", "version": 2, "status": "aligning", "summary": "减少重复检索", "objectives": ["提升命中率", "降低重复提问"], "scope": "知识库检索与问答", "in_scope": ["问答检索", "知识补充"], "out_of_scope": ["权限模型重构"], "success_criteria": ["首响 < 3s"], "constraints": ["附件仅作证据引用"], "milestones": ["需求冻结", "开发完成"], "risks": ["需求膨胀"], "signal_ids": [12], }, }, available_skills="FastAPI, React, LangChain", ) assert "需求宪章与项目上下文" in prompt assert "内部知识库升级宪章" in prompt assert "FastAPI, React, LangChain" in prompt assert "objectives" in prompt assert "constraints" in prompt @pytest.mark.asyncio async def test_general_task_breakdown_route_uses_project_and_charter_context(monkeypatch) -> None: captured: dict[str, object] = {} async def _fake_load_context(db, *, tenant_id: int, description: str, project_id=None, charter_id=None): _ = (db, tenant_id, project_id, charter_id) captured["description"] = description return { "request": {"description": description}, "project": { "id": 6, "title": "内部知识库升级", "description": "围绕内部知识库问答能力升级", "status": "active", "assigned_role": "backend", }, "charter": { "id": 55, "title": "内部知识库升级宪章", "version": 2, "status": "aligning", "summary": "减少重复检索", "objectives": ["提升命中率", "降低重复提问"], "scope": "知识库检索与问答", "in_scope": ["问答检索", "知识补充"], "out_of_scope": ["权限模型重构"], "success_criteria": ["首响 < 3s"], "constraints": ["附件仅作证据引用"], "milestones": ["需求冻结", "开发完成"], "risks": ["需求膨胀"], "signal_ids": [12], }, } async def _fake_load_skills(db, *, tenant_id: int): _ = (db, tenant_id) return "FastAPI, React, LangChain" class _FakeLLM: async def generate_structured_object(self, prompt: str, **kwargs): captured["prompt"] = prompt captured["llm_kwargs"] = kwargs payload = { "tasks": [ { "task_key": "task-1", "title": "需求澄清与范围确认", "description": "确认导出知识库的范围", "type": "ai", "estimated_hours": 4, "required_skill_names": ["FastAPI"], "required_skill_tag_ids": [101], "depends_on_task_keys": [], "risk_level": "low", "risk_notes": "边界明确", "confidence": 0.94, "evidence_refs": ["charter.objectives.0"], }, { "task_key": "task-2", "title": "核心方案落地", "description": "实现导出与验证流程", "type": "backend", "estimated_hours": 12, "required_skill_names": ["React", "LangChain"], "required_skill_tag_ids": [202, 303], "depends_on_task_keys": ["task-1"], "risk_level": "medium", "risk_notes": "依赖范围确认结果", "confidence": 0.87, "evidence_refs": ["charter.success_criteria.0"], }, ], "detected_skills": ["FastAPI", "React", "LangChain"], } return StructuredExtractionResult( payload=payload, source_field="message.content", provider="glm", model="qwen-plus", raw_text=json.dumps(payload, ensure_ascii=False), validation="ok", ) class _FakePlane: async def sync_tasks(self, tasks, is_urgent: bool = False): _ = is_urgent return {"synced_count": len(tasks)} monkeypatch.setattr(tasks_api, "_load_task_breakdown_context", _fake_load_context) monkeypatch.setattr(tasks_api, "_load_available_skill_names", _fake_load_skills) monkeypatch.setattr(tasks_api, "llm_service", _FakeLLM()) monkeypatch.setattr(tasks_api, "plane_service", _FakePlane()) result = await tasks_api.breakdown_tasks( TaskBreakdownRequest( description="支持导出知识库内容", project_id=6, charter_id=55, sync_to_plane=True, ), db=SimpleNamespace(), current_user=SimpleNamespace(tenant_id=1, id=1), ) assert "内部知识库升级宪章" in str(captured["prompt"]) assert "FastAPI, React, LangChain" in str(captured["prompt"]) assert result["sync_result"]["synced_count"] == 2 assert result["tasks"][0]["task_key"] == "task-1" assert result["tasks"][0]["required_skill_tag_ids"] == [101] assert result["tasks"][1]["depends_on_task_keys"] == ["task-1"] @pytest.mark.asyncio async def test_sstc_task_breakdown_route_persists_business_object_and_payload(monkeypatch) -> None: captured: dict[str, object] = {"drafts": []} async def _fake_load_context(db, *, tenant_id: int, description: str, project_id=None, charter_id=None): _ = (db, tenant_id, project_id, charter_id) captured["description"] = description return { "request": {"description": description}, "project": { "id": 6, "title": "内部知识库升级", "description": "围绕内部知识库问答能力升级", "status": "active", "assigned_role": "backend", }, "charter": { "id": 55, "title": "内部知识库升级宪章", "version": 2, "status": "aligning", "summary": "减少重复检索", "objectives": ["提升命中率", "降低重复提问"], "scope": "知识库检索与问答", "in_scope": ["问答检索", "知识补充"], "out_of_scope": ["权限模型重构"], "success_criteria": ["首响 < 3s"], "constraints": ["附件仅作证据引用"], "milestones": ["需求冻结", "开发完成"], "risks": ["需求膨胀"], "signal_ids": [12], }, } async def _fake_load_skills(db, *, tenant_id: int): _ = (db, tenant_id) return "FastAPI, React, LangChain" class _FakeLLM: async def generate_structured_object(self, prompt: str, **kwargs): captured["prompt"] = prompt captured["llm_kwargs"] = kwargs payload = { "tasks": [ { "task_key": "task-1", "title": "需求澄清与范围确认", "description": "确认导出知识库的范围", "type": "ai", "estimated_hours": 4, "required_skill_names": ["FastAPI"], "required_skill_tag_ids": [101], "depends_on_task_keys": [], "risk_level": "low", "risk_notes": "边界明确", "confidence": 0.94, "evidence_refs": ["charter.objectives.0"], }, { "task_key": "task-2", "title": "核心方案落地", "description": "实现导出与验证流程", "type": "backend", "estimated_hours": 12, "required_skill_names": ["React", "LangChain"], "required_skill_tag_ids": [202, 303], "depends_on_task_keys": ["task-1"], "risk_level": "medium", "risk_notes": "依赖范围确认结果", "confidence": 0.87, "evidence_refs": ["charter.success_criteria.0"], }, ], "detected_skills": ["FastAPI", "React", "LangChain"], } return StructuredExtractionResult( payload=payload, source_field="message.content", provider="glm", model="qwen-plus", raw_text=json.dumps(payload, ensure_ascii=False), validation="ok", ) class _FakeResult: def __init__(self, titles=None): self._titles = titles or [] def scalars(self): return self def all(self): return self._titles def first(self): return self._titles[0] if self._titles else None class _FakeDB: def __init__(self): self.added = [] self.committed = False async def execute(self, stmt): _ = stmt return _FakeResult([]) def add(self, value): self.added.append(value) async def commit(self): self.committed = True async def refresh(self, obj): _ = obj return None async def _fake_persist_task(db, **kwargs): _ = db draft = kwargs["draft"] captured["drafts"].append(draft) return SimpleNamespace( id=len(captured["drafts"]), title=draft.title, description=draft.description, priority=draft.priority, status=draft.status, assignee_id=draft.assignee_id, required_skill_tag_ids=list(draft.required_skill_tag_ids or []), ai_breakdown_payload=draft.ai_breakdown_payload, manual_overrides=draft.manual_overrides, ) async def _fake_ensure_project(service, project_id): _ = (service, project_id) return SimpleNamespace( id=6, title="内部知识库升级", name="内部知识库升级", description="围绕内部知识库问答能力升级", status="active", assigned_role="backend", ) async def _fake_ensure_charter(db, *, tenant_id: int, project_id: int, charter_id: int): _ = (db, tenant_id, project_id, charter_id) return SimpleNamespace( id=55, title="内部知识库升级宪章", version=2, content={"modules": {}}, status="aligning", ) def _fake_assert_charter_aligned(charter): _ = charter return None async def _fake_log_operation(*args, **kwargs): _ = (args, kwargs) return None async def _fake_invalidate_system_context(*args, **kwargs): _ = (args, kwargs) return None def _fake_sync_pmflow_rag_async(*args, **kwargs): _ = (args, kwargs) return None monkeypatch.setattr(sstc_tasks_ai_breakdown_api, "_ensure_project", _fake_ensure_project) monkeypatch.setattr(sstc_tasks_ai_breakdown_api, "_ensure_charter", _fake_ensure_charter) monkeypatch.setattr(sstc_tasks_ai_breakdown_api, "_assert_charter_aligned", _fake_assert_charter_aligned) monkeypatch.setattr(sstc_tasks_ai_breakdown_api, "_load_available_skill_names", _fake_load_skills) monkeypatch.setattr(sstc_tasks_ai_breakdown_api, "llm_service", _FakeLLM()) monkeypatch.setattr(sstc_tasks_ai_breakdown_api, "persist_task", _fake_persist_task) monkeypatch.setattr(sstc_tasks_ai_breakdown_api, "log_operation", _fake_log_operation) monkeypatch.setattr(sstc_tasks_ai_breakdown_api, "_sync_pmflow_rag_async", _fake_sync_pmflow_rag_async) monkeypatch.setattr(sstc_tasks_ai_breakdown_api, "context_cache_service", SimpleNamespace(invalidate_system_context=_fake_invalidate_system_context)) result = await sstc_tasks_ai_breakdown_api.ai_breakdown_tasks( project_id="6", charter_id="55", db=_FakeDB(), current_user=SimpleNamespace(tenant_id=1, id=1, role="REVIEWER"), permission_user=SimpleNamespace(), ) assert "内部知识库升级宪章" in str(captured["prompt"]) assert len(captured["drafts"]) == 2 assert captured["drafts"][0].required_skill_tag_ids == [101] assert captured["drafts"][0].ai_breakdown_payload["tasks"][0]["task_key"] == "task-1" assert captured["drafts"][0].manual_overrides == {} assert result["data"]["created_count"] == 2 assert result["data"]["items"][0]["title"] == "需求澄清与范围确认" @pytest.mark.asyncio async def test_recommend_assignee_route_uses_skill_tag_ids_semantically(monkeypatch) -> None: captured: dict[str, object] = {} class _FakeTask: id = 77 title = "导出知识库" description = "支持导出知识库内容" project_id = 6 required_skill_tag_ids = [101, 202] async def _fake_ensure_project(service, project_id): _ = (service, project_id) return None async def _fake_ensure_task(db, *, tenant_id: int, project_id: int, task_id: int): _ = (db, tenant_id, project_id, task_id) return _FakeTask() async def _fake_recommend_personnel(*, db, tenant_id: int, required_skills, task_description=None, task_id=None, limit=5): _ = (db, tenant_id, task_description, task_id, limit) captured["required_skills"] = list(required_skills) return { "task_id": task_id, "task_description": task_description, "required_skills": list(required_skills), "recommendations": [ { "user_id": "9", "user_name": "Ada", "match_score": 91, "reason": "strong match", "skill_match_count": 2, "avg_proficiency": 88, "workload_rate": 45, "recommended_role": "backend", } ], "total_count": 1, } class _FakeResult: def __init__(self, values): self._values = values def scalars(self): return self def all(self): return self._values async def _fake_ensure_project(service, project_id): _ = (service, project_id) return SimpleNamespace(id=6) async def _fake_ensure_task(db, *, tenant_id: int, project_id: int, task_id: int): _ = (db, tenant_id, project_id, task_id) return _FakeTask() class _FakeDB: async def execute(self, stmt): sql = str(stmt) if "user_projects" in sql.lower(): return _FakeResult([9]) return _FakeResult([9]) monkeypatch.setattr(sstc_tasks_recommend_optimize_api, "_ensure_project", _fake_ensure_project) monkeypatch.setattr(sstc_tasks_recommend_optimize_api, "_ensure_task", _fake_ensure_task) monkeypatch.setattr(sstc_tasks_recommend_optimize_api, "recommend_personnel", _fake_recommend_personnel) result = await sstc_tasks_recommend_optimize_api.recommend_assignee_for_task( project_id="6", task_id="77", db=_FakeDB(), current_user=SimpleNamespace(tenant_id=1, id=1), permission_user=SimpleNamespace(), ) assert captured["required_skills"] == [101, 202] assert result["data"]["recommendations"][0]["user_id"] == "9" assert result["data"]["total_count"] == 1