qiuruiandClaude Sonnet 4.6
df59fc0c83
包含从项目创建至今的全部代码首次入库： 核心框架： - TaskQueue（SQLite，6 种状态，原子 dequeue） - Executor（asyncio 并发，semaphore 限流，agent_role 路由） - Scheduler（APScheduler，三队列独立调度） - Watchdog（超时检测 + 优先级防饥饿） - Scanner（项目扫描，README/TODO/CLAUDE.md 提取） Agent 体系（20+ 角色）： - 项目交付组：architect/developer/tester/productizer - 基础技术组：base-architect/validator/hw/os/kernel/lowlevel/system-tester - 算法组：algo-antishake/position/nav - 洞察组：planner/vision-analyst/media-producer - 市场组：market-pm/sport/elder/safety - 组织层：boss/group-leader/senior-dev/ops - 平台扩展：nrf-dev/esp32-dev - 三专项调研：rtp-researcher/net-researcher/kernel-analyzer 工具层： - ProjectMemory（goals/facts/history SQLite） - VersionPipeline（并行版本流水线 + 反幻觉门控） - EventBus（Redis Streams，5 consumer groups） - DailyReport（append-only 日报 + Boss 决策视图） - AgentScorer（4 维评分：完成度/质量/自主性/协作） - DeviceAgent（SSH rsync + 板端执行） - EnvCollector（本机+SSH 环境采集） - ArchVersion（快照/优化/promote/rollback） Dashboard（React + Vite + Tailwind）： - 看板/项目/配置/洞察/日报/Visual 六标签页 - WebSocket 实时更新 - Markdown 渲染（react-markdown + remark-gfm） 测试：314 passed Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
80 lines
3.0 KiB
Python
80 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import pytest
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock
|
|
from nmfs_agents.agents.manager import ManagerAgent
|
|
from nmfs_agents.config import AgentsConfig, ProjectConfig, SchedulerConfig, ClaudeConfig, FeishuConfig
|
|
from nmfs_agents.core.queue import TaskQueue, Task
|
|
|
|
|
|
def _cfg(tmp_path: Path) -> AgentsConfig:
|
|
proj = tmp_path / "yolo"
|
|
proj.mkdir()
|
|
(proj / "README.md").write_text("## 已知问题\n- yolo12n 零检出\n")
|
|
return AgentsConfig(
|
|
projects={"yolo": ProjectConfig(path=proj, mode="report")},
|
|
scheduler=SchedulerConfig(max_concurrent=2),
|
|
claude=ClaudeConfig(api_key="fake"),
|
|
feishu=FeishuConfig(),
|
|
devices={},
|
|
)
|
|
|
|
|
|
def test_scan_and_enqueue_returns_count(tmp_path):
|
|
"""scan_and_enqueue 应返回新增任务数。"""
|
|
cfg = _cfg(tmp_path)
|
|
q = TaskQueue(db_path=tmp_path / "tasks.db")
|
|
manager = ManagerAgent(cfg, queue=q)
|
|
count = manager.scan_and_enqueue()
|
|
assert count >= 1
|
|
|
|
|
|
def test_get_status_returns_dict(tmp_path):
|
|
"""get_status 应返回各状态计数字典。"""
|
|
cfg = _cfg(tmp_path)
|
|
q = TaskQueue(db_path=tmp_path / "tasks.db")
|
|
q.enqueue(Task(project="yolo", type="fix_bug", title="test",
|
|
priority=1, mode="report", agent_role="developer"))
|
|
manager = ManagerAgent(cfg, queue=q)
|
|
status = manager.get_status()
|
|
assert "pending" in status
|
|
assert status["pending"] >= 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_cycle_executes_tasks(tmp_path):
|
|
"""run_cycle 应执行扫描+执行，返回统计。"""
|
|
cfg = _cfg(tmp_path)
|
|
q = TaskQueue(db_path=tmp_path / "tasks.db")
|
|
manager = ManagerAgent(cfg, queue=q)
|
|
|
|
mock_result = MagicMock(status="done", summary="done")
|
|
with patch("nmfs_agents.core.executor.DeveloperAgent") as MockAgent:
|
|
MockAgent.return_value.run.return_value = mock_result
|
|
stats = await manager.run_cycle()
|
|
|
|
assert stats["enqueued"] >= 1
|
|
assert stats["executed"] >= 1
|
|
|
|
|
|
def test_executor_confirm_mode_marks_done(tmp_path):
|
|
"""confirm 模式的任务执行后直接标记为 done（hook 级审批已在执行中发生）。"""
|
|
cfg = _cfg(tmp_path)
|
|
# 覆盖项目 mode 为 confirm
|
|
cfg.projects["yolo"] = ProjectConfig(path=tmp_path / "yolo", mode="confirm")
|
|
q = TaskQueue(db_path=tmp_path / "tasks.db")
|
|
tid = q.enqueue(Task(project="yolo", type="fix_bug", title="修复零检出",
|
|
priority=1, mode="confirm", agent_role="developer"))
|
|
|
|
mock_result = MagicMock(status="done", summary="diff: +sigmoid 修复")
|
|
with patch("nmfs_agents.core.executor.DeveloperAgent") as MockAgent:
|
|
MockAgent.return_value.run.return_value = mock_result
|
|
from nmfs_agents.core.executor import Executor
|
|
asyncio.run(Executor(cfg, queue=q).run_next())
|
|
|
|
with q._conn() as c:
|
|
row = c.execute("SELECT status FROM tasks WHERE id=?", (tid,)).fetchone()
|
|
assert row["status"] == "done"
|