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>
244 lines
10 KiB
Python
244 lines
10 KiB
Python
from __future__ import annotations
|
||
|
||
import pytest
|
||
from nmfs_agents.core.queue import TaskQueue, Task
|
||
|
||
|
||
def test_enqueue_and_dequeue(tmp_path):
|
||
q = TaskQueue(db_path=tmp_path / "tasks.db")
|
||
task_id = q.enqueue(Task(
|
||
project="yolo", type="fix_bug", title="修复 yolo12n 零检出",
|
||
priority=1, mode="confirm", agent_role="developer",
|
||
context="zero detection on board"
|
||
))
|
||
assert task_id > 0
|
||
next_task = q.dequeue()
|
||
assert next_task is not None
|
||
assert next_task.project == "yolo"
|
||
assert next_task.status == "running"
|
||
|
||
|
||
def test_priority_ordering(tmp_path):
|
||
q = TaskQueue(db_path=tmp_path / "tasks.db")
|
||
q.enqueue(Task(project="mediapipe", type="improve_perf", title="提升 FPS",
|
||
priority=3, mode="auto", agent_role="developer"))
|
||
q.enqueue(Task(project="yolo", type="fix_bug", title="修复零检出",
|
||
priority=1, mode="confirm", agent_role="developer"))
|
||
first = q.dequeue()
|
||
assert first.priority == 1 # 低数字 = 高优先级
|
||
|
||
|
||
def test_no_duplicate_pending(tmp_path):
|
||
q = TaskQueue(db_path=tmp_path / "tasks.db")
|
||
q.enqueue(Task(project="yolo", type="fix_bug", title="修复零检出",
|
||
priority=1, mode="confirm", agent_role="developer"))
|
||
# 同一 project+type+title 的任务已存在且 pending/running，不再重复入队
|
||
task_id = q.enqueue(Task(project="yolo", type="fix_bug", title="修复零检出",
|
||
priority=1, mode="confirm", agent_role="developer"))
|
||
assert task_id == -1 # -1 表示跳过
|
||
|
||
|
||
def test_mark_done(tmp_path):
|
||
q = TaskQueue(db_path=tmp_path / "tasks.db")
|
||
tid = q.enqueue(Task(project="yolo", type="fix_bug", title="修复",
|
||
priority=1, mode="auto", agent_role="developer"))
|
||
q.dequeue()
|
||
q.mark_done(tid, summary="修复成功")
|
||
assert q.dequeue() is None # 队列已空
|
||
|
||
|
||
def test_task_new_fields_exist():
|
||
"""Task 应有 retry_count/parent_task_id/constraint_type/awaiting_role 字段"""
|
||
t = Task(project="p", type="fix_bug", title="t",
|
||
priority=1, mode="auto", agent_role="developer")
|
||
assert t.retry_count == 0
|
||
assert t.parent_task_id == 0
|
||
assert t.constraint_type == ""
|
||
assert t.awaiting_role == ""
|
||
|
||
def test_enqueue_persists_new_fields(tmp_path):
|
||
q = TaskQueue(db_path=tmp_path / "t.db")
|
||
tid = q.enqueue(Task(project="p", type="fix_bug", title="t2",
|
||
priority=1, mode="auto", agent_role="developer",
|
||
retry_count=1, parent_task_id=10, constraint_type="veto_pending"))
|
||
with q._conn() as c:
|
||
row = c.execute("SELECT * FROM tasks WHERE id=?", (tid,)).fetchone()
|
||
assert row["retry_count"] == 1
|
||
assert row["parent_task_id"] == 10
|
||
assert row["constraint_type"] == "veto_pending"
|
||
|
||
def test_mark_waiting_approval(tmp_path):
|
||
q = TaskQueue(db_path=tmp_path / "t.db")
|
||
tid = q.enqueue(Task(project="p", type="fix_bug", title="t3",
|
||
priority=1, mode="auto", agent_role="developer"))
|
||
q.mark_waiting_approval(tid, awaiting_role="arch-security")
|
||
with q._conn() as c:
|
||
row = c.execute("SELECT status,awaiting_role FROM tasks WHERE id=?", (tid,)).fetchone()
|
||
assert row["status"] == "waiting_approval"
|
||
assert row["awaiting_role"] == "arch-security"
|
||
|
||
def test_mark_vetoed_and_requeue(tmp_path):
|
||
q = TaskQueue(db_path=tmp_path / "t.db")
|
||
tid = q.enqueue(Task(project="p", type="fix_bug", title="t4",
|
||
priority=1, mode="auto", agent_role="developer"))
|
||
# 先 dequeue（变 running），再 mark_vetoed
|
||
task = q.dequeue()
|
||
q.mark_vetoed(tid, reason="安全隐患")
|
||
# 重入队后应能再次 dequeue
|
||
task2 = q.dequeue()
|
||
assert task2 is not None
|
||
assert task2.retry_count == 1
|
||
assert "安全隐患" in task2.context
|
||
|
||
def test_dequeue_skips_waiting_approval(tmp_path):
|
||
q = TaskQueue(db_path=tmp_path / "t.db")
|
||
tid1 = q.enqueue(Task(project="p", type="fix_bug", title="t5",
|
||
priority=1, mode="auto", agent_role="developer"))
|
||
q.mark_waiting_approval(tid1, awaiting_role="pm")
|
||
tid2 = q.enqueue(Task(project="p", type="fix_bug", title="t6",
|
||
priority=2, mode="auto", agent_role="developer"))
|
||
task = q.dequeue()
|
||
assert task.id == tid2 # t5 在 waiting_approval，跳过
|
||
|
||
|
||
def test_mark_adversarial_pending(tmp_path):
|
||
q = TaskQueue(db_path=tmp_path / "t.db")
|
||
tid = q.enqueue(Task(project="p", type="fix_bug", title="提案",
|
||
priority=1, mode="auto", agent_role="arch-system"))
|
||
q.mark_adversarial_pending(tid)
|
||
with q._conn() as c:
|
||
row = c.execute("SELECT status,constraint_type FROM tasks WHERE id=?", (tid,)).fetchone()
|
||
assert row["status"] == "adversarial_pending"
|
||
assert row["constraint_type"] == "adversarial"
|
||
|
||
def test_approve_waiting(tmp_path):
|
||
q = TaskQueue(db_path=tmp_path / "t.db")
|
||
tid = q.enqueue(Task(project="p", type="fix_bug", title="等待审批",
|
||
priority=1, mode="auto", agent_role="developer"))
|
||
q.mark_waiting_approval(tid, awaiting_role="arch-security")
|
||
q.approve_waiting(tid)
|
||
with q._conn() as c:
|
||
row = c.execute("SELECT status,awaiting_role FROM tasks WHERE id=?", (tid,)).fetchone()
|
||
assert row["status"] == "done"
|
||
assert row["awaiting_role"] == ""
|
||
|
||
def test_get_waiting_approval_tasks(tmp_path):
|
||
q = TaskQueue(db_path=tmp_path / "t.db")
|
||
tid1 = q.enqueue(Task(project="p", type="fix_bug", title="w1",
|
||
priority=1, mode="auto", agent_role="developer"))
|
||
tid2 = q.enqueue(Task(project="p", type="fix_bug", title="w2",
|
||
priority=2, mode="auto", agent_role="developer"))
|
||
q.mark_waiting_approval(tid1, awaiting_role="pm")
|
||
# tid2 仍为 pending，不在 waiting_approval 列表
|
||
tasks = q.get_waiting_approval_tasks()
|
||
assert len(tasks) == 1
|
||
assert tasks[0].id == tid1
|
||
assert tasks[0].awaiting_role == "pm"
|
||
|
||
|
||
def test_reset_stale_running_all(tmp_path):
|
||
"""timeout_minutes=0 重置所有 running 任务"""
|
||
q = TaskQueue(db_path=tmp_path / "t.db")
|
||
tid1 = q.enqueue(Task(project="p", type="fix_bug", title="r1",
|
||
priority=1, mode="auto", agent_role="developer"))
|
||
tid2 = q.enqueue(Task(project="p", type="fix_bug", title="r2",
|
||
priority=2, mode="auto", agent_role="developer"))
|
||
q.dequeue()
|
||
q.dequeue()
|
||
count = q.reset_stale_running(timeout_minutes=0)
|
||
assert count == 2
|
||
with q._conn() as c:
|
||
rows = c.execute("SELECT status FROM tasks WHERE id IN (?,?)", (tid1, tid2)).fetchall()
|
||
assert all(r["status"] == "failed" for r in rows)
|
||
|
||
|
||
def test_reset_stale_running_timeout(tmp_path):
|
||
"""running 时间未超时的任务不被重置"""
|
||
q = TaskQueue(db_path=tmp_path / "t.db")
|
||
tid = q.enqueue(Task(project="p", type="fix_bug", title="r3",
|
||
priority=1, mode="auto", agent_role="developer"))
|
||
q.dequeue()
|
||
# 超时阈值设为 9999 分钟，刚启动的任务不应被重置
|
||
count = q.reset_stale_running(timeout_minutes=9999)
|
||
assert count == 0
|
||
with q._conn() as c:
|
||
row = c.execute("SELECT status FROM tasks WHERE id=?", (tid,)).fetchone()
|
||
assert row["status"] == "running"
|
||
|
||
|
||
def test_bump_stale_priorities(tmp_path):
|
||
"""长期 pending 任务优先级应被提升（数值减 1）"""
|
||
from datetime import datetime, timedelta
|
||
q = TaskQueue(db_path=tmp_path / "t.db")
|
||
tid = q.enqueue(Task(project="p", type="fix_bug", title="aging",
|
||
priority=3, mode="auto", agent_role="developer"))
|
||
# 伪造 created_at 为 3 小时前
|
||
old_time = (datetime.utcnow() - timedelta(hours=3)).isoformat()
|
||
with q._conn() as c:
|
||
c.execute("UPDATE tasks SET created_at=? WHERE id=?", (old_time, tid))
|
||
count = q.bump_stale_priorities(threshold_minutes=120)
|
||
assert count == 1
|
||
with q._conn() as c:
|
||
row = c.execute("SELECT priority FROM tasks WHERE id=?", (tid,)).fetchone()
|
||
assert row["priority"] == 2 # 3 → 2
|
||
|
||
|
||
def test_bump_stale_priorities_no_below_one(tmp_path):
|
||
"""优先级为 1 的任务不继续提升"""
|
||
from datetime import datetime, timedelta
|
||
q = TaskQueue(db_path=tmp_path / "t.db")
|
||
tid = q.enqueue(Task(project="p", type="fix_bug", title="max_pri",
|
||
priority=1, mode="auto", agent_role="developer"))
|
||
old_time = (datetime.utcnow() - timedelta(hours=3)).isoformat()
|
||
with q._conn() as c:
|
||
c.execute("UPDATE tasks SET created_at=? WHERE id=?", (old_time, tid))
|
||
count = q.bump_stale_priorities(threshold_minutes=120)
|
||
assert count == 0 # priority=1 不能再提升
|
||
|
||
|
||
def test_mark_vetoed_exceeds_limit(tmp_path):
|
||
"""连续 3 次否决后任务状态变为 failed（超出最大重试次数）"""
|
||
q = TaskQueue(db_path=tmp_path / "t.db")
|
||
tid = q.enqueue(Task(project="p", type="fix_bug", title="超限测试",
|
||
priority=1, mode="auto", agent_role="developer"))
|
||
# 第 1 次否决
|
||
q.dequeue() # 变为 running
|
||
q.mark_vetoed(tid, reason="第1次")
|
||
# 第 2 次否决
|
||
q.dequeue() # 再次变为 running
|
||
q.mark_vetoed(tid, reason="第2次")
|
||
# 第 3 次否决 → 应变为 failed
|
||
q.dequeue()
|
||
q.mark_vetoed(tid, reason="第3次超限")
|
||
with q._conn() as c:
|
||
row = c.execute("SELECT status,result_summary FROM tasks WHERE id=?", (tid,)).fetchone()
|
||
assert row["status"] == "failed"
|
||
assert "超限" in row["result_summary"]
|
||
|
||
|
||
def test_dequeue_with_role_filter(tmp_path):
|
||
"""dequeue 支持 role_filter，只取特定角色的任务"""
|
||
q = TaskQueue(db_path=tmp_path / "t.db")
|
||
q.enqueue(Task(project="a", type="fix", title="base task",
|
||
context="", priority=3, mode="auto", agent_role="os-engineer"))
|
||
q.enqueue(Task(project="b", type="fix", title="project task",
|
||
context="", priority=3, mode="auto", agent_role="developer"))
|
||
|
||
base_roles = frozenset({"os-engineer", "algo-researcher"})
|
||
t = q.dequeue(role_filter=base_roles)
|
||
assert t is not None
|
||
assert t.agent_role == "os-engineer"
|
||
|
||
t2 = q.dequeue(role_filter=base_roles)
|
||
assert t2 is None # developer 任务不在 base_roles 中
|
||
|
||
|
||
def test_dequeue_role_filter_none_returns_any(tmp_path):
|
||
"""role_filter=None 时行为与原来相同（取所有角色）"""
|
||
q = TaskQueue(db_path=tmp_path / "t.db")
|
||
q.enqueue(Task(project="a", type="fix", title="t1",
|
||
context="", priority=3, mode="auto", agent_role="developer"))
|
||
t = q.dequeue(role_filter=None)
|
||
assert t is not None
|
||
assert t.agent_role == "developer"
|