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>
68 lines
2.3 KiB
Python
68 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
import pytest
|
|
from nmfs_agents.core.queue import TaskQueue, Task
|
|
from nmfs_agents.core.watchdog import Watchdog
|
|
|
|
|
|
def _make_task(title: str, priority: int = 3) -> Task:
|
|
return Task(project="p", type="fix_bug", title=title,
|
|
priority=priority, mode="auto", agent_role="developer")
|
|
|
|
|
|
def test_watchdog_starts_and_stops(tmp_path):
|
|
"""Watchdog 线程可正常启动和停止"""
|
|
q = TaskQueue(db_path=tmp_path / "t.db")
|
|
wd = Watchdog(q, interval=1)
|
|
wd.start()
|
|
assert wd.is_alive()
|
|
wd.stop()
|
|
wd.join(timeout=3)
|
|
assert not wd.is_alive()
|
|
|
|
|
|
def test_watchdog_resets_stale_tasks(tmp_path):
|
|
"""Watchdog tick 将超时任务标记为 failed"""
|
|
from datetime import datetime, timedelta
|
|
q = TaskQueue(db_path=tmp_path / "t.db")
|
|
tid = q.enqueue(_make_task("stale_task"))
|
|
q.dequeue() # 变为 running
|
|
# 伪造 started_at 为 3 小时前（超过默认 2h 阈值）
|
|
old_time = (datetime.utcnow() - timedelta(hours=3)).isoformat()
|
|
with q._conn() as c:
|
|
c.execute("UPDATE tasks SET started_at=? WHERE id=?", (old_time, tid))
|
|
|
|
# 用 1 分钟超时触发（测试中用 1 分钟阈值）
|
|
wd = Watchdog(q, interval=1, task_timeout_minutes=1)
|
|
wd.start()
|
|
time.sleep(2) # 等 watchdog 至少执行一次 tick
|
|
wd.stop()
|
|
wd.join(timeout=3)
|
|
|
|
with q._conn() as c:
|
|
row = c.execute("SELECT status FROM tasks WHERE id=?", (tid,)).fetchone()
|
|
assert row["status"] == "failed"
|
|
|
|
|
|
def test_watchdog_ages_priorities(tmp_path):
|
|
"""Watchdog tick 提升长期 pending 任务优先级"""
|
|
from datetime import datetime, timedelta
|
|
q = TaskQueue(db_path=tmp_path / "t.db")
|
|
tid = q.enqueue(_make_task("old_task", priority=3))
|
|
# 伪造 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))
|
|
|
|
wd = Watchdog(q, interval=1, task_timeout_minutes=9999)
|
|
wd.start()
|
|
time.sleep(2)
|
|
wd.stop()
|
|
wd.join(timeout=3)
|
|
|
|
with q._conn() as c:
|
|
row = c.execute("SELECT priority FROM tasks WHERE id=?", (tid,)).fetchone()
|
|
assert row["priority"] < 3 # 至少提升了一次（3 → 2 或更低，取决于 tick 次数）
|