auto-sync: 2026-05-17 06:00:34

tests/test_dispatcher.py

@@ -0,0 +1,246 @@
+"""F9 Agent 调度器单元测试
+
+按 test-plan-v2.6.md §F9：
+- T1: 三级决策树（P0）
+- T2: 调度不阻塞（P0）
+- T3: 队列满拒绝（P0）
+- T4: 任务优先级排序（P1）
+"""
+
+import asyncio
+import json
+import pytest
+from pathlib import Path
+from typing import Any, Dict, Optional
+from unittest.mock import AsyncMock, MagicMock
+
+from src.blackboard.models import Task
+from src.daemon.dispatcher import Dispatcher, DispatchLevel
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def dispatcher():
+    return Dispatcher(
+        registered_agents=["zhangfei-dev", "guanyu-dev", "simayi-challenger"],
+    )
+
+
+@pytest.fixture
+def task_pending():
+    return Task(
+        id="t1", title="Write code", status="pending",
+        assigned_by="daemon", task_type="coding",
+        assignee="zhangfei-dev",
+    )
+
+
+@pytest.fixture
+def task_no_assignee():
+    return Task(
+        id="t2", title="Generic task", status="pending",
+        assigned_by="daemon",
+    )
+
+
+@pytest.fixture
+def task_unknown_agent():
+    return Task(
+        id="t3", title="Unknown agent task", status="pending",
+        assigned_by="daemon", assignee="nonexistent-agent",
+    )
+
+
+# ---------------------------------------------------------------------------
+# T1: 三级决策树
+# ---------------------------------------------------------------------------
+
+class TestDecisionTree:
+    def test_local_action(self, dispatcher, task_pending):
+        """L1 本地执行"""
+        decision = dispatcher.decide(task_pending, "L1_guardrail")
+        assert decision["level"] == DispatchLevel.LOCAL
+        assert decision["agent_id"] == "daemon"
+
+    def test_local_format_check(self, dispatcher, task_pending):
+        """format_check → 本地"""
+        decision = dispatcher.decide(task_pending, "format_check")
+        assert decision["level"] == DispatchLevel.LOCAL
+
+    def test_local_file_exists(self, dispatcher, task_pending):
+        """file_exists_check → 本地"""
+        decision = dispatcher.decide(task_pending, "file_exists_check")
+        assert decision["level"] == DispatchLevel.LOCAL
+
+    def test_registered_agent(self, dispatcher, task_pending):
+        """注册 Agent → Full Agent"""
+        decision = dispatcher.decide(task_pending)
+        assert decision["level"] == DispatchLevel.FULL_AGENT
+        assert decision["agent_id"] == "zhangfei-dev"
+
+    def test_adjudication_new_session(self, dispatcher, task_pending):
+        """adjudication → new_session=True"""
+        decision = dispatcher.decide(task_pending, "adjudication")
+        assert decision["level"] == DispatchLevel.FULL_AGENT
+        assert decision["new_session"] is True
+
+    def test_no_assignee_subagent(self, dispatcher, task_no_assignee):
+        """无 assignee → Subagent"""
+        decision = dispatcher.decide(task_no_assignee)
+        assert decision["level"] == DispatchLevel.SUB_AGENT
+
+    def test_unknown_agent_escalate(self, dispatcher, task_unknown_agent):
+        """未注册 Agent → 升级庞统"""
+        decision = dispatcher.decide(task_unknown_agent)
+        assert decision["level"] == DispatchLevel.ESCALATE
+        assert decision["agent_id"] == "pangtong-fujunshi"
+        assert decision["new_session"] is True
+
+    def test_different_registered_agents(self, dispatcher):
+        """不同注册 Agent 正确路由"""
+        for agent_id in ["zhangfei-dev", "guanyu-dev", "simayi-challenger"]:
+            task = Task(id="t", title="T", status="pending",
+                        assigned_by="d", assignee=agent_id)
+            decision = dispatcher.decide(task)
+            assert decision["level"] == DispatchLevel.FULL_AGENT
+            assert decision["agent_id"] == agent_id
+
+    def test_unknown_action_with_registered_agent(self, dispatcher, task_pending):
+        """未知 action_type 但有注册 assignee → Full Agent"""
+        decision = dispatcher.decide(task_pending, "some_unknown_action")
+        assert decision["level"] == DispatchLevel.FULL_AGENT
+
+
+# ---------------------------------------------------------------------------
+# T2: 调度执行（带 mock）
+# ---------------------------------------------------------------------------
+
+class TestDispatch:
+    def test_dispatch_local(self, dispatcher, task_pending):
+        """本地调度不需要 spawner"""
+        result = asyncio.run(dispatcher.dispatch(task_pending, "L1_guardrail"))
+        assert result["status"] == "dispatched"
+        assert result["level"] == "local"
+
+    def test_dispatch_full_agent(self, dispatcher, task_pending):
+        """Full Agent 调度"""
+        mock_spawner = MagicMock()
+        mock_spawner.spawn_full_agent = AsyncMock(return_value="session-123")
+        dispatcher.spawner = mock_spawner
+
+        result = asyncio.run(dispatcher.dispatch(task_pending))
+        assert result["status"] == "dispatched"
+        assert result["session_id"] == "session-123"
+        assert result["level"] == "full_agent"
+
+    def test_dispatch_no_spawner_returns_error(self, dispatcher, task_pending):
+        """无 spawner → error"""
+        result = asyncio.run(dispatcher.dispatch(task_pending))
+        assert result["status"] == "error"
+        assert "No spawner" in result["reason"]
+
+    def test_dispatch_subagent(self, dispatcher, task_no_assignee):
+        """Subagent 调度"""
+        mock_spawner = MagicMock()
+        mock_spawner.spawn_subagent = AsyncMock(return_value="sub-123")
+        dispatcher.spawner = mock_spawner
+
+        result = asyncio.run(dispatcher.dispatch(task_no_assignee))
+        assert result["status"] == "dispatched"
+        assert result["level"] == "sub_agent"
+
+    def test_dispatch_escalate(self, dispatcher, task_unknown_agent):
+        """升级调度"""
+        mock_spawner = MagicMock()
+        mock_spawner.spawn_full_agent = AsyncMock(return_value="esc-123")
+        dispatcher.spawner = mock_spawner
+
+        result = asyncio.run(dispatcher.dispatch(task_unknown_agent))
+        assert result["status"] == "dispatched"
+        assert result["level"] == "escalate"
+
+    def test_dispatch_spawn_failure(self, dispatcher, task_pending):
+        """spawn 失败 → error"""
+        mock_spawner = MagicMock()
+        mock_spawner.spawn_full_agent = AsyncMock(side_effect=RuntimeError("spawn failed"))
+        dispatcher.spawner = mock_spawner
+
+        result = asyncio.run(dispatcher.dispatch(task_pending))
+        assert result["status"] == "error"
+
+
+# ---------------------------------------------------------------------------
+# T3: 队列满拒绝
+# ---------------------------------------------------------------------------
+
+class TestConcurrencyControl:
+    def test_counter_busy_skips(self, dispatcher, task_pending):
+        """Agent 忙 → skip"""
+        mock_counter = MagicMock()
+        mock_counter.can_acquire = AsyncMock(return_value=False)
+        dispatcher.counter = mock_counter
+
+        result = asyncio.run(dispatcher.dispatch(task_pending))
+        assert result["status"] == "skipped"
+        assert "busy" in result["reason"].lower()
+
+    def test_counter_releases_on_error(self, dispatcher, task_pending):
+        """spawn 失败后释放 counter"""
+        mock_spawner = MagicMock()
+        mock_spawner.spawn_full_agent = AsyncMock(side_effect=RuntimeError("fail"))
+        mock_counter = MagicMock()
+        mock_counter.can_acquire = AsyncMock(return_value=True)
+        mock_counter.acquire = AsyncMock()
+        mock_counter.release = MagicMock()
+        dispatcher.spawner = mock_spawner
+        dispatcher.counter = mock_counter
+
+        result = asyncio.run(dispatcher.dispatch(task_pending))
+        assert result["status"] == "error"
+        mock_counter.release.assert_called_once_with("zhangfei-dev")
+
+    def test_local_not_blocked_by_counter(self, dispatcher, task_pending):
+        """本地执行不受 counter 限制"""
+        mock_counter = MagicMock()
+        mock_counter.can_acquire = AsyncMock(return_value=False)
+        dispatcher.counter = mock_counter
+
+        result = asyncio.run(dispatcher.dispatch(task_pending, "L1_guardrail"))
+        assert result["status"] == "dispatched"
+
+
+# ---------------------------------------------------------------------------
+# T4: 批量决策
+# ---------------------------------------------------------------------------
+
+class TestBatchDecision:
+    def test_dispatch_pending(self, dispatcher):
+        """批量决策返回所有任务的决策"""
+        tasks = [
+            Task(id="t1", title="T1", status="pending",
+                 assigned_by="d", assignee="zhangfei-dev"),
+            Task(id="t2", title="T2", status="pending",
+                 assigned_by="d", assignee=None),
+            Task(id="t3", title="T3", status="pending",
+                 assigned_by="d", assignee="unknown"),
+        ]
+        results = dispatcher.dispatch_pending(tasks)
+        assert len(results) == 3
+        assert results[0]["level"] == DispatchLevel.FULL_AGENT
+        assert results[1]["level"] == DispatchLevel.SUB_AGENT
+        assert results[2]["level"] == DispatchLevel.ESCALATE
+
+    def test_message_building(self, dispatcher):
+        """构建给 Agent 的消息"""
+        task = Task(
+            id="t1", title="Build Feature", status="pending",
+            assigned_by="daemon", task_type="coding",
+            description="Implement X", must_haves="Tests, Docs",
+        )
+        msg = dispatcher._build_message(task, "execute")
+        assert "Build Feature" in msg
+        assert "Implement X" in msg
+        assert "execute" in msg