import os import unittest from datetime import datetime from tempfile import NamedTemporaryFile from unittest.mock import AsyncMock, patch from sqlalchemy import and_, select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from app.models import Base, Budget, PortfolioSnapshot, Project, ProjectDomain, Task, TruthNode, User, UserProject, UserSkillBinding, UserWorkload, WorkItem from app.services.ai_pmo_cadence_service import AIPMOCadenceService from app.services.ai_pmo_kernel_service import ( AIPMOKernelService, INTENT_SIGNAL_SOURCE_ONBOARD, IntentResolver, ) from app.services.ai_pmo_domain_workflow import SignalSourceDomainWorkflow from app.services.signal_source_orchestration_service import SignalSourceOrchestrationService class AIPMOKernelServiceTest(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self) -> None: tmp = NamedTemporaryFile(prefix="pmflow-ai-pmo-test-", suffix=".db", delete=False) tmp.close() self.db_path = tmp.name self.engine = create_async_engine(f"sqlite+aiosqlite:///{self.db_path}", future=True) self.session_factory = async_sessionmaker(self.engine, expire_on_commit=False, class_=AsyncSession) async with self.engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) async with self.session_factory() as db: user = User( tenant_id=1, username="boss", email="boss@example.com", password_hash="hash", name="老板", role="BOSS", assigned_domains=[], workload=0, status="active", is_active=True, ) db.add(user) await db.flush() domain = ProjectDomain( tenant_id=1, name="医药", slug="pharma", description="医药研究域", config={}, is_active=True, is_default=False, signal_count=0, context_node_count=0, experiment_count=0, owner_id=str(user.id), ) db.add(domain) await db.flush() project = Project( tenant_id=1, name="医药行业十五五规划布局研究", title="医药行业十五五规划布局研究", description="研究医药行业方向和机会。", owner_id=int(user.id), status="active", domain_id=int(domain.id), assigned_role="pm", signal_count=0, ) db.add(project) await db.flush() charter = TruthNode( tenant_id=1, node_type="charter", project_id=int(project.id), signal_id=None, created_by=int(user.id), title="项目宪章", content={"success_criteria": ["两周内形成研究报告"]}, version=1, status="active", ) db.add(charter) await db.flush() db.add( Budget( tenant_id=1, budget_type="project", project_id=int(project.id), period="monthly", budget_amount_usd=1000.0, alert_threshold_percent=80.0, hard_limit_enabled=True, current_usage_usd=500.0, current_period_start=datetime(2026, 3, 1), current_period_end=datetime(2026, 3, 31), status="active", override_history={}, ) ) db.add(UserProject(tenant_id=1, user_id=int(user.id), project_id=int(project.id))) db.add( UserWorkload( tenant_id=1, user_id=int(user.id), current_assigned_hours=20.0, max_available_hours_per_week=40.0, workload_ratio=0.5, ) ) db.add( UserSkillBinding( tenant_id=1, user_id=int(user.id), skill_tag_id=1, proficiency_level="expert", proficiency=5, level="expert", years_experience=8.0, quality_score=9.0, efficiency_score=8.5, ) ) db.add( Task( tenant_id=1, truth_node_id=int(charter.id), project_id=int(project.id), assignee_id=int(user.id), ai_recommended_assignees=None, title="梳理外部依赖", description="当前依赖较多，任务处于 blocked。", required_skill_tag_ids=[2], status="blocked", priority="high", estimated_hours=8.0, actual_hours=0.0, quality_score=None, due_date=datetime(2026, 3, 1), completed_at=None, ) ) await db.commit() async def asyncTearDown(self) -> None: await self.engine.dispose() if getattr(self, "db_path", None): try: os.remove(self.db_path) except FileNotFoundError: pass def test_intent_resolver_detects_signal_source_request(self) -> None: intent = IntentResolver.resolve( text="给项目ID: 1 新增一个 GitHub 信号源，监控 openai/codex 仓库。", project_id=1, ) self.assertEqual(intent.name, INTENT_SIGNAL_SOURCE_ONBOARD) async def test_build_world_model_includes_budget_capacity_and_skill_graph(self) -> None: async with self.session_factory() as db: service = AIPMOKernelService(db, tenant_id=1) world_model = await service._build_world_model( message_text="给项目ID: 1 新增一个医药信号。", sender="老板", project_hint_id=1, ) self.assertEqual(world_model["budget"]["budget_id"], 1) self.assertEqual(len(world_model["capacity"]), 1) self.assertEqual(len(world_model["skill_graph"]), 1) self.assertEqual(world_model["resolved_project_id"], 1) self.assertEqual(world_model["project_domain"]["slug"], "pharma") self.assertEqual(world_model["dependencies"]["pressure"], "high") self.assertEqual(world_model["task_summary"]["uncovered_skill_tag_ids"], [2]) async def test_process_query_message_persists_work_item_and_reply_receipt(self) -> None: async with self.session_factory() as db: service = AIPMOKernelService(db, tenant_id=1) with patch( "app.services.ai_pmo_kernel_service.AdaptiveRAGService.answer_question", AsyncMock(return_value={"answer": "当前项目状态正常。", "sources": []}), ): result = await service.process_ingress( source_system="pmflow_manual", source_message_id="msg-query-1", message_text="项目ID: 1 现在状态如何？", sender="老板", channel_ref="pmflow", project_hint_id=1, ) work_item = ( await db.execute( select(WorkItem).where( and_( WorkItem.tenant_id == 1, WorkItem.source_message_id == "msg-query-1", ) ) ) ).scalar_one() review_cycle = ( await db.execute( select(PortfolioSnapshot).where( and_( PortfolioSnapshot.tenant_id == 1, PortfolioSnapshot.scope_type == "review_cycle", PortfolioSnapshot.scope_ref == f"work_item:{int(work_item.id)}", ) ) ) ).scalar_one() self.assertTrue(result["handled"]) self.assertEqual(result["status"], "answered") self.assertTrue(result["direct_reply_required"]) self.assertEqual(result["direct_reply_text"], "当前项目状态正常。") self.assertEqual(result["decision_level"], "project") self.assertIn("agent_quality_score", work_item.review_summary) self.assertTrue(bool(work_item.review_summary.get("verified"))) self.assertEqual(review_cycle.summary["work_item_id"], int(work_item.id)) async def test_duplicate_message_is_filtered(self) -> None: async with self.session_factory() as db: service = AIPMOKernelService(db, tenant_id=1) with patch( "app.services.ai_pmo_kernel_service.AdaptiveRAGService.answer_question", AsyncMock(return_value={"answer": "OK", "sources": []}), ): first = await service.process_ingress( source_system="pmflow_manual", source_message_id="dup-1", message_text="项目ID: 1 现在状态如何？", sender="老板", channel_ref="pmflow", project_hint_id=1, ) second = await service.process_ingress( source_system="pmflow_manual", source_message_id="dup-1", message_text="项目ID: 1 现在状态如何？", sender="老板", channel_ref="pmflow", project_hint_id=1, ) self.assertEqual(first["status"], "answered") self.assertEqual(second["status"], "ignored") self.assertFalse(second["direct_reply_required"]) async def test_policy_gate_blocks_archive_like_request(self) -> None: async with self.session_factory() as db: service = AIPMOKernelService(db, tenant_id=1) result = await service.process_ingress( source_system="pmflow_manual", source_message_id="gate-1", message_text="请把项目ID: 1 直接归档。", sender="老板", channel_ref="pmflow", project_hint_id=1, ) self.assertTrue(result["handled"]) self.assertEqual(result["status"], "captured") self.assertIn("人工确认", result["direct_reply_text"]) async def test_process_project_create_uses_new_kernel(self) -> None: async with self.session_factory() as db: service = AIPMOKernelService(db, tenant_id=1) with patch( "app.services.ai_pmo_kernel_service.AIStewardService.create_project_from_natural_language", AsyncMock( return_value={ "project": {"id": 9, "title": "医药新项目", "name": "医药新项目"}, "signal": {"id": 11}, "tasks": [{"id": 21, "title": "任务A"}], } ), ): result = await service.process_ingress( source_system="pmflow_manual", source_message_id="project-1", message_text="请创建项目：医药新项目，做医药趋势研究。", sender="老板", channel_ref="pmflow", ) work_item = ( await db.execute( select(WorkItem).where( and_( WorkItem.tenant_id == 1, WorkItem.source_message_id == "project-1", ) ) ) ).scalar_one() self.assertEqual(result["status"], "captured") self.assertEqual(result["created_project_id"], 9) self.assertEqual(result["linked_project_ids"], [9]) self.assertTrue(result["direct_reply_required"]) self.assertEqual(work_item.project_id, 9) async def test_signal_source_request_hits_simulation_policy_gate_when_skill_gap_exists(self) -> None: async with self.session_factory() as db: service = AIPMOKernelService(db, tenant_id=1) result = await service.process_ingress( source_system="pmflow_manual", source_message_id="source-1", message_text="给项目ID: 1 新增一个 GitHub 信号源，监控 openai/codex。", sender="老板", channel_ref="pmflow", project_hint_id=1, ) self.assertTrue(result["handled"]) self.assertEqual(result["status"], "captured") self.assertIn("技能缺口", result["direct_reply_text"]) async def test_signal_create_stays_auto_executable_under_dependency_pressure(self) -> None: async with self.session_factory() as db: service = AIPMOKernelService(db, tenant_id=1) result = await service.process_ingress( source_system="pmflow_manual", source_message_id="signal-auto-1", message_text="给项目ID: 1 新增一条信号：跟踪江苏医保支付改革和器械招采拐点。", sender="老板", channel_ref="pmflow", project_hint_id=1, ) self.assertTrue(result["handled"]) self.assertEqual(result["status"], "captured") self.assertIsNotNone(result["signal_id"]) self.assertEqual(result["linked_project_ids"], [1]) self.assertIn("已入库", result["direct_reply_text"]) async def test_final_phase_generates_risk_cadence_and_program_view(self) -> None: async with self.session_factory() as db: service = AIPMOKernelService(db, tenant_id=1) result = await service.process_ingress( source_system="pmflow_manual", source_message_id="source-risk-1", message_text="给项目ID: 1 新增一个 GitHub 信号源，监控 openai/codex。", sender="老板", channel_ref="pmflow", project_hint_id=1, ) cadence_service = AIPMOCadenceService(db, tenant_id=1) latest_risk = await cadence_service.get_latest_snapshot( cadence_type="risk_review", project_id=1, ) program_view = await cadence_service.build_program_view() self.assertTrue(result["handled"]) self.assertEqual(result["status"], "captured") self.assertIsNotNone(latest_risk) self.assertEqual(latest_risk["scope_type"], "risk_review") self.assertGreaterEqual(int(latest_risk["summary"]["blocked_task_count"]), 1) self.assertEqual(program_view["summary"]["program_count"], 1) self.assertGreaterEqual(int(program_view["summary"]["pending_manual_total"]), 1) async def test_followup_continue_resumes_single_waiting_manual_work_item(self) -> None: async with self.session_factory() as db: service = AIPMOKernelService(db, tenant_id=1) first = await service.process_ingress( source_system="feishu", source_message_id="feishu-source-1", message_text="给项目ID: 1 新增一个 GitHub 信号源，监控 openai/codex。", sender="老板", channel_ref="oc_phase3", source_sender_open_id="ou_boss", project_hint_id=1, ) with patch( "app.services.ai_pmo_kernel_service.SignalSourceDomainWorkflow.run", AsyncMock( return_value={ "status": "completed", "registry_id": 7, "typed_actions": [{"operation_code": "task.create", "status": "applied"}], "material_action_count": 1, "fallback_used": False, "raw_result": {}, } ), ): followup = await service.process_ingress( source_system="feishu", source_message_id="feishu-followup-1", message_text="继续处理", sender="老板", channel_ref="oc_phase3", source_sender_open_id="ou_boss", project_hint_id=1, ) original = ( await db.execute( select(WorkItem).where( and_( WorkItem.tenant_id == 1, WorkItem.source_message_id == "feishu-source-1", ) ) ) ).scalar_one() self.assertEqual(first["status"], "captured") self.assertEqual(followup["status"], "answered") self.assertIn("已继续处理", followup["direct_reply_text"]) self.assertEqual(original.state, "succeeded") async def test_followup_continue_requires_explicit_target_when_multiple_waiting_manual_items(self) -> None: async with self.session_factory() as db: service = AIPMOKernelService(db, tenant_id=1) await service.process_ingress( source_system="feishu", source_message_id="feishu-source-2a", message_text="给项目ID: 1 新增一个 GitHub 信号源，监控 openai/codex。", sender="老板", channel_ref="oc_phase3_multi", source_sender_open_id="ou_boss", project_hint_id=1, ) await service.process_ingress( source_system="feishu", source_message_id="feishu-source-2b", message_text="给项目ID: 1 新增一个 Web 搜索信号源，监控医保动态。", sender="老板", channel_ref="oc_phase3_multi", source_sender_open_id="ou_boss", project_hint_id=1, ) followup = await service.process_ingress( source_system="feishu", source_message_id="feishu-followup-2", message_text="继续处理", sender="老板", channel_ref="oc_phase3_multi", source_sender_open_id="ou_boss", project_hint_id=1, ) followup_item = ( await db.execute( select(WorkItem).where( and_( WorkItem.tenant_id == 1, WorkItem.source_message_id == "feishu-followup-2", ) ) ) ).scalar_one() self.assertEqual(followup["status"], "captured") self.assertIn("多个待确认工作项", followup["direct_reply_text"]) self.assertEqual(followup_item.state, "waiting_manual") async def test_followup_explicit_work_item_id_can_resume_across_source_system(self) -> None: async with self.session_factory() as db: service = AIPMOKernelService(db, tenant_id=1) first = await service.process_ingress( source_system="feishu", source_message_id="feishu-source-3", message_text="给项目ID: 1 新增一个 GitHub 信号源，监控 openai/codex。", sender="老板", channel_ref="oc_phase3_explicit", source_sender_open_id="ou_boss", project_hint_id=1, ) target_id = int(first["work_item_id"]) with patch( "app.services.ai_pmo_kernel_service.SignalSourceDomainWorkflow.run", AsyncMock( return_value={ "status": "completed", "registry_id": 9, "typed_actions": [{"operation_code": "task.create", "status": "applied"}], "material_action_count": 1, "fallback_used": False, "raw_result": {}, } ), ): followup = await service.process_ingress( source_system="ai_pmo_ui", source_message_id="resume-ui-1", message_text=f"继续处理 work_item #{target_id}", sender="老板", channel_ref="ai_pmo_ui", project_hint_id=1, ) self.assertEqual(followup["status"], "answered") self.assertIn("已继续处理", followup["direct_reply_text"]) def test_signal_source_domain_workflow_keeps_notification_only_pending_manual(self) -> None: normalized = SignalSourceDomainWorkflow.normalize( { "status": "partial", "signal_source": { "registry_id": 7, "status": "pending", "onboarding_status": "pending_manual", }, "openclaw": { "status": "completed", "result": { "execution": { "results": [ { "operation_code": "notification.create", "status": "applied", "payload": {"title": "待确认"}, } ] } }, }, "worker_executions": [ { "worker_role_key": "pm_dispatcher", "status": "completed", "used_fallback": True, } ], "pending_workers": [], } ) self.assertEqual(normalized["status"], "pending_manual") self.assertEqual(normalized["material_action_count"], 0) async def test_transport_mode_statement_is_answered_without_creating_work_item(self) -> None: async with self.session_factory() as db: service = AIPMOKernelService(db, tenant_id=1) result = await service.process_ingress( source_system="pmflow_manual", source_message_id="transport-mode-1", message_text="现在都是以长连接的形式，并不是webhook。", sender="老板", channel_ref="pmflow", project_hint_id=1, ) persisted = ( await db.execute( select(WorkItem).where( and_( WorkItem.tenant_id == 1, WorkItem.source_message_id == "transport-mode-1", ) ) ) ).scalar_one_or_none() self.assertTrue(result["handled"]) self.assertEqual(result["status"], "answered") self.assertTrue(result["direct_reply_required"]) self.assertIn("长连接模式", result["direct_reply_text"]) self.assertIsNone(persisted) async def test_signal_source_registry_write_re_normalizes_invalid_kind_and_generic_name(self) -> None: async with self.session_factory() as db: service = SignalSourceOrchestrationService(db) result = await service._create_or_reuse_registry( tenant_id=1, project_id=1, signal_id=None, operator_user_id=None, requested_via="feishu", source_payload={ "source_kind": "long", "source_name": "给项目ID", "source_identifier": "long:给项目id", "description": "每日搜寻最新的医药行业变革动态", "config": {"keywords": ["医药行业", "变革"]}, }, initiating_text="给项目ID: 1 接一个复合信号源：每日搜寻最新的医药行业变革动态。", dry_run=False, ) self.assertEqual(result["source_kind"], "feishu") self.assertNotEqual(result["source_name"], "给项目ID") self.assertTrue(str(result["source_identifier"]).startswith("feishu:"))