nmfs-agents/docs/plans/2026-03-08-team-agents-impl.md

三组 15 角色团队架构 Implementation Plan

For Claude: REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.

Goal: 将现有单队列 8 角色扩展为三组 15 角色架构（基础技术研发组 / 项目研发交付组 / 产品洞察与市场落地组），实现独立队列调度、测试能力大幅增强、视觉分析和市场文档能力。

Architecture: 三个独立 SQLite 任务队列（base_opt/project_delivery/insight），各自 Executor 实例和并发限制，通过 ROLE_TO_QUEUE 映射路由。新 Agent 均继承 DeveloperAgent，通过 system prompt 注入专业领域知识和 skills 配置。

Tech Stack: Python 3.10, SQLite, APScheduler, claude subprocess, Playwright MCP, pytest

---

Task 1: Config 扩展 — 多队列配置支持

Files:

• Modify: src/rockchip_agents/config.py
• Modify: configs/agents.yaml
• Test: tests/test_config.py

Step 1: 写失败测试

在 tests/test_config.py 末尾追加：

def test_queue_configs_loaded(tmp_path):
    """AgentsConfig 能加载三组队列配置"""
    from rockchip_agents.config import AgentsConfig, QueueConfig
    cfg = AgentsConfig.__new__(AgentsConfig)
    base_q = QueueConfig(
        db="data/base_opt_tasks.db",
        max_concurrent=3,
        scan_interval_hours=4,
        roles=frozenset({"hw-engineer", "os-engineer", "algo-researcher",
                         "base-validator", "base-architect"}),
    )
    assert base_q.max_concurrent == 3
    assert "hw-engineer" in base_q.roles

Step 2: 运行确认失败

cd /data/rockchip/agents && source venv/bin/activate
pytest tests/test_config.py::test_queue_configs_loaded -v

Step 3: 在 config.py 中新增 QueueConfig dataclass（在 SchedulerConfig 之前）

@dataclass
class QueueConfig:
    db: str                          # 相对于项目根目录的 DB 路径
    max_concurrent: int = 3
    scan_interval_hours: int = 1
    roles: frozenset[str] = field(default_factory=frozenset)

在 AgentsConfig 中新增字段（第 59 行之后）：

@dataclass
class AgentsConfig:
    projects: dict[str, ProjectConfig]
    scheduler: SchedulerConfig
    claude: ClaudeConfig
    feishu: FeishuConfig
    devices: dict[str, DeviceConfig]
    claude_wx_url: str = ""
    queues: dict[str, QueueConfig] = field(default_factory=dict)  # ← 新增

Step 4: 修改 load_config 解析队列配置

在 load_config 函数末尾（return 之前）添加队列解析：

# 解析多队列配置
queues: dict[str, QueueConfig] = {}
for q_name, q_data in agent_data.get("queues", {}).items():
    queues[q_name] = QueueConfig(
        db=q_data.get("db", f"data/{q_name}_tasks.db"),
        max_concurrent=q_data.get("max_concurrent", 3),
        scan_interval_hours=q_data.get("scan_interval_hours", 1),
        roles=frozenset(q_data.get("roles", [])),
    )

并在 AgentsConfig(...) 构造处传入 queues=queues。

Step 5: 修改 configs/agents.yaml 新增队列配置

在文件末尾追加：

queues:
  base_opt:
    db: data/base_opt_tasks.db
    max_concurrent: 3
    scan_interval_hours: 4
    roles:
      - hw-engineer
      - os-engineer
      - algo-researcher
      - base-validator
      - base-architect

  project_delivery:
    db: data/tasks.db
    max_concurrent: 3
    scan_interval_hours: 1
    roles:
      - developer
      - dev-kernel
      - dev-lowlevel
      - system-tester
      - architect

  insight:
    db: data/insight_tasks.db
    max_concurrent: 2
    scan_interval_hours: 6
    roles:
      - planner
      - productizer
      - vision-analyst
      - market-pm
      - media-producer

Step 6: 运行测试

pytest tests/test_config.py -v

---

Task 2: 队列路由工具 — queue_for_role + mcp_tools 更新

Files:

• Modify: src/rockchip_agents/core/queue.py
• Modify: configs/mcp_tools.yaml
• Test: tests/test_queue.py

Step 1: 写失败测试

在 tests/test_queue.py 末尾追加：

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_opt 组的任务
    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"

    # 再取，应该是 None（只有 developer 剩余）
    t2 = q.dequeue(role_filter=base_roles)
    assert t2 is None

Step 2: 运行确认失败

pytest tests/test_queue.py::test_dequeue_with_role_filter -v

Step 3: 修改 queue.py — dequeue 支持 role_filter

找到 dequeue 方法（约第 108 行），在参数中新增 role_filter：

def dequeue(
    self,
    exclude_projects: frozenset[str] | None = None,
    role_filter: frozenset[str] | None = None,
) -> Optional[Task]:

在方法内部，SQL 查询部分增加 role_filter 条件：

# 构建 WHERE 子句
conditions = ["status='pending'"]
params: list = []

if exclude_projects:
    placeholders = ",".join("?" * len(exclude_projects))
    conditions.append(f"project NOT IN ({placeholders})")
    params.extend(tuple(exclude_projects))

if role_filter:
    placeholders = ",".join("?" * len(role_filter))
    conditions.append(f"agent_role IN ({placeholders})")
    params.extend(tuple(role_filter))

where = " AND ".join(conditions)
row = conn.execute(
    f"SELECT * FROM tasks WHERE {where} ORDER BY priority ASC, id ASC LIMIT 1",
    params,
).fetchone()

（删除原有的两段 if/else SQL 写法，统一用上面的构建方式）

Step 4: 更新 configs/mcp_tools.yaml — 新增角色 skills

在 roles: 下追加：

  os-engineer:      [sequential-thinking, fetch]
  algo-researcher:  [sequential-thinking, fetch]
  base-validator:   [sequential-thinking]
  base-architect:   [sequential-thinking, fetch]
  system-tester:    [sequential-thinking, fetch, playwright]
  vision-analyst:   [sequential-thinking, playwright]
  market-pm:        [sequential-thinking, fetch]
  media-producer:   [sequential-thinking]

Step 5: 运行测试

pytest tests/test_queue.py -v

---

Task 3: 多队列 Executor + Scheduler

Files:

• Modify: src/rockchip_agents/core/executor.py
• Modify: src/rockchip_agents/core/scheduler.py
• Test: tests/test_executor.py

Step 1: 修改 executor.py — run_all_pending 支持 role_filter

在 run_all_pending 方法签名增加 role_filter 参数：

async def run_all_pending(
    self,
    skip_projects: frozenset[str] | None = None,
    role_filter: frozenset[str] | None = None,
) -> int:

在方法内的 dequeue 调用改为：

task = self._queue.dequeue(
    exclude_projects=skip_projects,
    role_filter=role_filter,
)

Step 2: 修改 scheduler.py — 三队列调度

将 _run_pending 函数改为带 queue_name 参数：

def _run_pending_queue(queue_name: str) -> None:
    """执行指定队列的待处理任务。"""
    from pathlib import Path
    cfg = load_config()
    q_cfg = cfg.queues.get(queue_name)
    if not q_cfg:
        logger.warning("队列配置不存在: %s", queue_name)
        return

    db_path = Path(q_cfg.db)
    from rockchip_agents.core.queue import TaskQueue
    queue = TaskQueue(db_path=db_path)
    executor = Executor(cfg, queue=queue)
    role_filter = q_cfg.roles if q_cfg.roles else None
    count = asyncio.run(executor.run_all_pending(role_filter=role_filter))
    logger.info("[%s] 本轮执行任务数: %d", queue_name, count)

    # insight 队列空时触发 planner
    if count == 0 and queue_name == "project_delivery":
        _enqueue_planner_task()
        logger.info("project_delivery 队列为空，已触发 PlannerAgent")

在 start() 函数中，替换原有的单 _run_pending job，改为三个队列各自的 job：

def start() -> None:
    cfg = load_config()

    # 启动恢复
    for q_name, q_cfg in cfg.queues.items():
        from pathlib import Path
        from rockchip_agents.core.queue import TaskQueue
        q = TaskQueue(db_path=Path(q_cfg.db))
        recovered = q.reset_stale_running(timeout_minutes=0)
        if recovered:
            logger.warning("[%s] 启动恢复: %d 个遗留任务重置", q_name, recovered)

    watchdog = Watchdog(TaskQueue())
    watchdog.start()

    scheduler = BlockingScheduler()
    # 主扫描（project_delivery，1h）
    scheduler.add_job(_scan_and_enqueue, "interval", hours=1, id="scan")
    # 三队列执行 job
    scheduler.add_job(
        lambda: _run_pending_queue("project_delivery"),
        "interval", hours=1, id="run_project",
        next_run_time=datetime.now() + timedelta(minutes=5),
    )
    scheduler.add_job(
        lambda: _run_pending_queue("base_opt"),
        "interval", hours=4, id="run_base",
        next_run_time=datetime.now() + timedelta(minutes=15),
    )
    scheduler.add_job(
        lambda: _run_pending_queue("insight"),
        "interval", hours=6, id="run_insight",
        next_run_time=datetime.now() + timedelta(minutes=10),
    )
    scheduler.add_job(
        _enqueue_planner_task, "interval", hours=6, id="planner",
        next_run_time=datetime.now() + timedelta(minutes=10),
    )

    logger.info("Scheduler 已启动，三队列模式")
    _scan_and_enqueue()
    try:
        scheduler.start()
    finally:
        watchdog.stop()

Step 3: 运行全量测试

pytest tests/ -q 2>&1 | tail -10

---

Task 4: base-validator — 基础技术验证 Agent

Files:

• Create: src/rockchip_agents/agents/base_validator.py
• Test: tests/test_base_validator.py

Step 1: 写失败测试

创建 tests/test_base_validator.py：

from __future__ import annotations
import pytest
from unittest.mock import MagicMock
from rockchip_agents.agents.base_validator import BaseValidatorAgent, BASE_VALIDATOR_SYSTEM
from rockchip_agents.config import AgentsConfig, ClaudeConfig
from rockchip_agents.core.queue import Task
from rockchip_agents.tools.project_memory import ProjectMemory


def _make_config():
    cfg = MagicMock(spec=AgentsConfig)
    cfg.projects = {"yolo": MagicMock(description="目标检测", mode="auto",
                                       path=MagicMock())}
    cfg.claude = ClaudeConfig(model="claude-sonnet-4-6", api_key=None)
    return cfg


def test_base_validator_system_prompt_contains_keywords():
    """系统提示包含性能、稳定性、精度等关键词"""
    assert "性能" in BASE_VALIDATOR_SYSTEM
    assert "稳定性" in BASE_VALIDATOR_SYSTEM
    assert "精度" in BASE_VALIDATOR_SYSTEM
    assert "硬件约束" in BASE_VALIDATOR_SYSTEM


def test_base_validator_context_contains_hardware_facts(tmp_path):
    """context 中包含 __hardware__ facts"""
    mem = ProjectMemory(db_path=tmp_path / "m.db")
    mem.set_fact("__hardware__", "rk3588.npu", "6TOPS", role="hw-engineer")
    cfg = _make_config()

    agent = BaseValidatorAgent(cfg, memory=mem)
    task = Task(project="yolo", type="validate", title="test",
                context="", priority=3, mode="auto", agent_role="base-validator")
    ctx = agent._build_context(task)
    assert "6TOPS" in ctx or "rk3588.npu" in ctx

Step 2: 运行确认失败

pytest tests/test_base_validator.py -v 2>&1 | head -15

Step 3: 创建 base_validator.py

from __future__ import annotations

from rockchip_agents.agents.developer import DeveloperAgent
from rockchip_agents.config import AgentsConfig
from rockchip_agents.core.queue import Task
from rockchip_agents.tools.project_memory import ProjectMemory
from typing import Optional

BASE_VALIDATOR_SYSTEM = """你是基础技术研发组的验证工程师，负责算法和OS的基线验证。

验证维度：
1. 精度测试：检出率、误检率、漏检率，与基线对比
2. 性能测试：FPS、推理延迟（ms）、NPU利用率、CPU/内存占用
3. 稳定性测试：长时间运行（≥1h）、内存泄漏检测、异常恢复
4. 硬件约束验证：读取 __hardware__ facts，确认算法满足硬件限制

输出格式：
- 验证结论（通过/失败/需优化）
- 性能指标数值（写入能力事实）
- [记忆] __hardware__: key=value（更新能力事实）
- [需求] <项目>: 具体优化需求（仅当发现问题时）

请用中文输出，数值精确，结论明确。
"""


class BaseValidatorAgent(DeveloperAgent):
    """基础技术组：算法精度 + 性能 + 稳定性 + 硬件约束验证"""

    def __init__(
        self,
        config: AgentsConfig,
        memory: Optional[ProjectMemory] = None,
    ) -> None:
        super().__init__(config, memory=memory)

    def _build_context(self, task: Task) -> str:
        base = super()._build_context(task)
        # 注入 __hardware__ facts
        hw_facts = self._memory.get_facts("__hardware__")
        if hw_facts:
            hw_lines = "\n".join(f"  {k}: {v}" for k, v in hw_facts.items())
            hw_section = f"\n【硬件约束（来自 hw-engineer）】\n{hw_lines}\n"
        else:
            hw_section = "\n【硬件约束】暂无，请查阅 __hardware__ facts\n"
        return base + hw_section + BASE_VALIDATOR_SYSTEM

Step 4: 运行测试

pytest tests/test_base_validator.py -v

---

Task 5: system-tester — 项目交付组增强测试 Agent

Files:

• Create: src/rockchip_agents/agents/system_tester.py
• Test: tests/test_system_tester.py

Step 1: 写失败测试

创建 tests/test_system_tester.py：

from __future__ import annotations
import pytest
from unittest.mock import MagicMock
from rockchip_agents.agents.system_tester import SystemTesterAgent, SYSTEM_TESTER_SYSTEM
from rockchip_agents.config import AgentsConfig, ClaudeConfig
from rockchip_agents.core.queue import Task


def _make_config():
    cfg = MagicMock(spec=AgentsConfig)
    cfg.projects = {"yolo": MagicMock(description="目标检测", mode="auto",
                                       path=MagicMock())}
    cfg.claude = ClaudeConfig(model="claude-sonnet-4-6", api_key=None)
    return cfg


def test_system_tester_system_prompt_covers_all_layers():
    """系统提示覆盖功能/视觉/性能/跨平台/报告五层"""
    for keyword in ["功能测试", "视觉测试", "性能测试", "跨平台", "测试报告"]:
        assert keyword in SYSTEM_TESTER_SYSTEM, f"缺少关键词: {keyword}"


def test_system_tester_context_has_playwright_hint(tmp_path):
    """context 中包含 Playwright 使用提示"""
    from rockchip_agents.tools.project_memory import ProjectMemory
    mem = ProjectMemory(db_path=tmp_path / "m.db")
    agent = SystemTesterAgent(_make_config(), memory=mem)
    task = Task(project="yolo", type="test", title="系统测试",
                context="", priority=2, mode="auto", agent_role="system-tester")
    ctx = agent._build_context(task)
    assert "Playwright" in ctx or "playwright" in ctx

Step 2: 运行确认失败

pytest tests/test_system_tester.py -v 2>&1 | head -15

Step 3: 创建 system_tester.py

from __future__ import annotations

from rockchip_agents.agents.developer import DeveloperAgent
from rockchip_agents.config import AgentsConfig
from rockchip_agents.core.queue import Task
from rockchip_agents.tools.project_memory import ProjectMemory
from typing import Optional

SYSTEM_TESTER_SYSTEM = """你是项目研发与交付组的系统测试工程师。

测试五层能力：
1. 功能测试：运行 pytest，SSH 板端自动执行，检查通过/失败
2. 视觉测试：使用 Playwright 截图推理界面，分析检测框/FPS/误检，截图保存至 docs/test_reports/
3. 性能测试：测量 FPS、推理延迟、内存占用，与目标值对比，写入 [记忆]
4. 跨平台测试：RK3588/全志/ESP32 分平台调度执行，记录差异
5. 测试报告：生成结构化 Markdown 报告至 docs/test_reports/YYYY-MM-DD-{project}.md

工具使用：
- Playwright：browser_navigate → 推理演示页 → browser_take_screenshot → 分析截图
- SSH 板端：通过 Bash 执行 ssh user@host "pytest tests/ -q"
- 性能测量：time/perf_counter 或板端 top/free 命令

失败处理：
- 测试失败 → 输出 [需求] <项目>: 具体问题描述，触发 developer 修复
- 性能退化 → [记忆] <项目>: fps_baseline=N，记录当前基线

请用中文，数值精确，测试结论明确。
"""


class SystemTesterAgent(DeveloperAgent):
    """项目交付组：多层系统测试（功能+视觉+性能+跨平台+报告）"""

    def __init__(
        self,
        config: AgentsConfig,
        memory: Optional[ProjectMemory] = None,
    ) -> None:
        super().__init__(config, memory=memory)

    def _build_context(self, task: Task) -> str:
        base = super()._build_context(task)
        return base + "\n" + SYSTEM_TESTER_SYSTEM

Step 4: 运行测试

pytest tests/test_system_tester.py -v

---

Task 6: 基础技术研发组三个新 Agent

Files:

• Create: src/rockchip_agents/agents/os_engineer.py
• Create: src/rockchip_agents/agents/algo_researcher.py
• Create: src/rockchip_agents/agents/base_architect.py
• Test: tests/test_base_tech_agents.py

Step 1: 写失败测试

创建 tests/test_base_tech_agents.py：

from __future__ import annotations
import pytest
from unittest.mock import MagicMock
from rockchip_agents.config import AgentsConfig, ClaudeConfig
from rockchip_agents.core.queue import Task
from rockchip_agents.tools.project_memory import ProjectMemory


def _make_config():
    cfg = MagicMock(spec=AgentsConfig)
    cfg.projects = {}
    cfg.claude = ClaudeConfig(model="claude-sonnet-4-6", api_key=None)
    return cfg


def _make_task(role):
    return Task(project="base", type="research", title="test",
                context="", priority=3, mode="auto", agent_role=role)


def test_os_engineer_context_contains_rtos_and_linux(tmp_path):
    from rockchip_agents.agents.os_engineer import OsEngineerAgent
    mem = ProjectMemory(db_path=tmp_path / "m.db")
    agent = OsEngineerAgent(_make_config(), memory=mem)
    ctx = agent._build_context(_make_task("os-engineer"))
    assert "RTOS" in ctx
    assert "Linux" in ctx or "内核" in ctx


def test_algo_researcher_context_contains_algorithm_domains(tmp_path):
    from rockchip_agents.agents.algo_researcher import AlgoResearcherAgent
    mem = ProjectMemory(db_path=tmp_path / "m.db")
    agent = AlgoResearcherAgent(_make_config(), memory=mem)
    ctx = agent._build_context(_make_task("algo-researcher"))
    assert "防抖" in ctx
    assert "惯导" in ctx or "GPS" in ctx


def test_base_architect_context_focuses_on_platform(tmp_path):
    from rockchip_agents.agents.base_architect import BaseArchitectAgent
    mem = ProjectMemory(db_path=tmp_path / "m.db")
    agent = BaseArchitectAgent(_make_config(), memory=mem)
    ctx = agent._build_context(_make_task("base-architect"))
    assert "平台" in ctx or "基础技术" in ctx

Step 2: 运行确认失败

pytest tests/test_base_tech_agents.py -v 2>&1 | head -15

Step 3: 创建 os_engineer.py

from __future__ import annotations
from rockchip_agents.agents.developer import DeveloperAgent
from rockchip_agents.core.queue import Task

OS_ENGINEER_SYSTEM = """你是基础技术研发组的 OS 工程师，专注 RTOS 和 Linux 基础层。

职责范围：
- RTOS：FreeRTOS/RT-Thread 任务调度、中断优先级、驱动框架、实时性验证
- Linux：内核 Kconfig 模板、DTS 框架、驱动基础层、内存管理
- Bootloader：U-Boot 基础配置、启动时序、分区管理
- 双栈协作：Linux/RTOS 在同一平台（如 RK3588 AMP 模式）的协调

输出能力事实格式：
- [记忆] __hardware__: os.rtos.latency_us=N, os.linux.boot_ms=N

工程规范：
- 地址空间严格区分物理/虚拟地址
- RTOS 任务优先级：中断 > 实时 > 普通
- 修改 DTS 前必须了解上下游节点
"""


class OsEngineerAgent(DeveloperAgent):
    def _build_context(self, task: Task) -> str:
        return super()._build_context(task) + "\n" + OS_ENGINEER_SYSTEM

Step 4: 创建 algo_researcher.py

from __future__ import annotations
from rockchip_agents.agents.developer import DeveloperAgent
from rockchip_agents.core.queue import Task
from rockchip_agents.config import AgentsConfig
from rockchip_agents.tools.project_memory import ProjectMemory
from typing import Optional

ALGO_RESEARCHER_SYSTEM = """你是基础技术研发组的算法研究员，专注嵌入式平台算法研究与落地。

研究方向：
- 防抖算法：陀螺仪数据融合、EIS（电子防抖）、OIS 协同
- 短距定位：UWB/BLE RSSI 测距、ToF 测距、多点定位融合
- GPS/惯导：IMU（加速度计+陀螺仪）融合、Kalman 滤波、姿态解算

硬件约束意识：
- 始终读取 __hardware__ facts 确认目标平台算力/内存
- TinyML 算法须在 ESP32/Nordic 上验证（资源极度受限）
- RKNN 算法在 NPU 上验证推理效率

输出能力事实格式：
- [记忆] __hardware__: algo.antishake.fps=N, algo.gps_ins.latency_ms=N
"""


class AlgoResearcherAgent(DeveloperAgent):
    def __init__(self, config: AgentsConfig,
                 memory: Optional[ProjectMemory] = None) -> None:
        super().__init__(config, memory=memory)

    def _build_context(self, task: Task) -> str:
        base = super()._build_context(task)
        # 注入 __hardware__ 约束
        hw_facts = self._memory.get_facts("__hardware__")
        hw_section = ""
        if hw_facts:
            lines = "\n".join(f"  {k}: {v}" for k, v in hw_facts.items())
            hw_section = f"\n【硬件约束】\n{lines}\n"
        return base + hw_section + "\n" + ALGO_RESEARCHER_SYSTEM

Step 5: 创建 base_architect.py

from __future__ import annotations
from rockchip_agents.agents.architect import ArchitectAgent
from rockchip_agents.core.queue import Task

BASE_ARCHITECT_SYSTEM = """你是基础技术研发组的架构师，专注平台级技术栈的整体设计。

职责（区别于项目组 architect）：
- 分析 OS/算法/硬件 三个方向的技术债和瓶颈
- 识别跨方向整合机会（如防抖算法与 GPS 惯导融合）
- 评估新硬件对现有基础能力的影响
- 为 os-engineer / algo-researcher 提出具体 [优化] 任务

输出格式：
- [优化] os-engineer: 具体优化建议
- [优化] algo-researcher: 具体优化建议
- [记忆] __platform__: key=value（记录架构决策）
"""


class BaseArchitectAgent(ArchitectAgent):
    def _build_prompt(self, summaries: dict[str, str]) -> str:
        base = super()._build_prompt(summaries)
        return BASE_ARCHITECT_SYSTEM + "\n\n" + base

Step 6: 运行测试

pytest tests/test_base_tech_agents.py -v
pytest tests/ -q 2>&1 | tail -5

---

Task 7: vision-analyst — Playwright + LLM 视觉分析 Agent

Files:

• Create: src/rockchip_agents/agents/vision_analyst.py
• Test: tests/test_vision_analyst.py

Step 1: 写失败测试

创建 tests/test_vision_analyst.py：

from __future__ import annotations
import pytest
from unittest.mock import MagicMock
from rockchip_agents.agents.vision_analyst import VisionAnalystAgent, VISION_ANALYST_SYSTEM
from rockchip_agents.config import AgentsConfig, ClaudeConfig
from rockchip_agents.core.queue import Task, TaskQueue
from rockchip_agents.tools.project_memory import ProjectMemory


def _make_config():
    cfg = MagicMock(spec=AgentsConfig)
    cfg.projects = {"yolo": MagicMock(description="目标检测", mode="auto",
                                       path=MagicMock())}
    cfg.claude = ClaudeConfig(model="claude-sonnet-4-6", api_key=None)
    return cfg


def test_vision_analyst_system_prompt_has_playwright_workflow():
    """提示词包含 Playwright 工作流关键词"""
    for kw in ["Playwright", "截图", "LLM", "检测框"]:
        assert kw in VISION_ANALYST_SYSTEM, f"缺少: {kw}"


def test_vision_analyst_extract_needs(tmp_path):
    """从视觉分析报告中提取 [需求] 任务"""
    mem = ProjectMemory(db_path=tmp_path / "m.db")
    q = TaskQueue(db_path=tmp_path / "t.db")
    agent = VisionAnalystAgent(_make_config(), queue=q, memory=mem)

    report = "视觉分析结果：检测框偏移明显\n- [需求] yolo: 修正检测框坐标偏移\n"
    agent._extract_needs(report)

    conn = q._conn()
    rows = conn.execute("SELECT title FROM tasks").fetchall()
    conn.close()
    assert any("检测框坐标偏移" in r["title"] for r in rows)

Step 2: 运行确认失败

pytest tests/test_vision_analyst.py -v 2>&1 | head -15

Step 3: 创建 vision_analyst.py

from __future__ import annotations

import logging
import os
import re
from typing import Optional

from rockchip_agents.agents.developer import (
    AgentResult, _make_drop_root_preexec, _run_with_log, _task_log_path,
)
from rockchip_agents.config import AgentsConfig
from rockchip_agents.core.queue import Task, TaskQueue
from rockchip_agents.tools.project_memory import ProjectMemory

logger = logging.getLogger(__name__)
TIMEOUT = 600

VISION_ANALYST_SYSTEM = """你是产品洞察与市场落地组的视觉分析师。

工作流程：
1. 使用 Playwright 导航至板端推理演示页面或监控大屏
2. browser_take_screenshot 截图（保存至 docs/screenshots/）
3. 作为多模态 LLM，直接分析截图内容：
   - 检测框位置是否合理（有无明显偏移/错位）
   - FPS 数值读取与目标值对比
   - 漏检情况（应检测目标未出现框）
   - 误检情况（背景被误识别）
4. 结合推理日志（JSON/txt）做定量验证
5. 输出分析报告 + 改进需求

输出格式：
- [需求] <项目名>: 具体视觉问题描述和改进方向
- 性能数值（FPS、检出率等）

请用中文，视觉描述精确，问题定位清晰。
"""


class VisionAnalystAgent:
    def __init__(
        self,
        config: AgentsConfig,
        queue: Optional[TaskQueue] = None,
        memory: Optional[ProjectMemory] = None,
    ) -> None:
        self._config = config
        self._queue = queue or TaskQueue()
        self._memory = memory or ProjectMemory()

    def run(self, task: Task) -> AgentResult:
        from rockchip_agents.config import load_mcp_tools
        mcp_cfg = load_mcp_tools()
        registry = mcp_cfg.get("registry", {})

        import shutil, json, tempfile
        mcp_servers = {}
        for tool in ["playwright", "sequential-thinking"]:
            if tool in registry:
                parts = registry[tool].split()
                if shutil.which(parts[0]):
                    mcp_servers[tool] = {"command": parts[0], "args": parts[1:]}

        tmp_files = []
        cmd = [
            "claude", "--print",
            "--output-format", "json",
            "--no-session-persistence",
            "--model", self._config.claude.model,
            "--dangerously-skip-permissions",
            "--append-system-prompt", VISION_ANALYST_SYSTEM,
        ]
        if mcp_servers:
            with tempfile.NamedTemporaryFile(
                mode="w", suffix=".json", delete=False, prefix="mcp_vision_"
            ) as f:
                json.dump({"mcpServers": mcp_servers}, f)
                tmp_files.append(f.name)
            cmd += ["--mcp-config", tmp_files[-1]]
        cmd += ["--", self._build_prompt(task)]

        try:
            env = self._build_env()
            result = _run_with_log(
                cmd, None, env, TIMEOUT, _task_log_path(task.id),
                preexec_fn=_make_drop_root_preexec(env),
            )
            if result.status == "done":
                self._extract_needs(result.summary)
            return result
        except Exception as e:
            logger.error("VisionAnalystAgent 失败: %s", e)
            return AgentResult(status="failed", summary=str(e))
        finally:
            for f in tmp_files:
                try:
                    os.unlink(f)
                except OSError:
                    pass

    def _build_prompt(self, task: Task) -> str:
        proj_cfg = self._config.projects.get(task.project)
        desc = proj_cfg.description if proj_cfg else task.project
        return (
            f"项目: {task.project}，描述: {desc}\n"
            f"任务: {task.title}\n"
            f"背景: {task.context}\n\n"
            f"请使用 Playwright 截图并分析推理效果，输出视觉分析报告。"
        )

    def _extract_needs(self, report_text: str) -> None:
        """从视觉分析报告提取 [需求] 任务入队"""
        for m in re.finditer(r"\[需求\]\s+([\w-]+)\s*[：:]\s*(.+)", report_text):
            project = m.group(1).strip()
            desc = m.group(2).strip()[:120]
            if project not in self._config.projects:
                continue
            proj_cfg = self._config.projects[project]
            self._queue.enqueue(Task(
                project=project, type="fix_bug",
                title=f"[视觉分析] {desc}",
                context=f"由 VisionAnalystAgent 发现\n{desc}",
                priority=2, mode=proj_cfg.mode,
                agent_role="developer", initiator="vision-analyst",
            ))

    def _build_env(self) -> dict:
        env = os.environ.copy()
        if self._config.claude.api_key:
            env["ANTHROPIC_API_KEY"] = self._config.claude.api_key
        env.pop("CLAUDECODE", None)
        env.pop("CLAUDE_CODE_ENTRYPOINT", None)
        return env

Step 4: 运行测试

pytest tests/test_vision_analyst.py -v

---

Task 8: market-pm + media-producer

Files:

• Create: src/rockchip_agents/agents/market_pm.py
• Create: src/rockchip_agents/agents/media_producer.py
• Test: tests/test_market_agents.py

Step 1: 写失败测试

from __future__ import annotations
import pytest
from unittest.mock import MagicMock
from rockchip_agents.config import AgentsConfig, ClaudeConfig
from rockchip_agents.core.queue import Task
from rockchip_agents.tools.project_memory import ProjectMemory


def _make_config():
    cfg = MagicMock(spec=AgentsConfig)
    cfg.projects = {}
    cfg.claude = ClaudeConfig(model="claude-sonnet-4-6", api_key=None)
    return cfg


def test_market_pm_context_contains_market_domain(tmp_path):
    from rockchip_agents.agents.market_pm import MarketPmAgent, MARKET_PM_SYSTEM
    assert "方案书" in MARKET_PM_SYSTEM or "白皮书" in MARKET_PM_SYSTEM
    mem = ProjectMemory(db_path=tmp_path / "m.db")
    agent = MarketPmAgent(_make_config(), memory=mem)
    task = Task(project="yolo", type="market", title="方案书",
                context="", priority=3, mode="report", agent_role="market-pm")
    ctx = agent._build_context(task)
    assert len(ctx) > 10


def test_media_producer_context_contains_tools(tmp_path):
    from rockchip_agents.agents.media_producer import MediaProducerAgent, MEDIA_PRODUCER_SYSTEM
    assert "OpenCV" in MEDIA_PRODUCER_SYSTEM or "ffmpeg" in MEDIA_PRODUCER_SYSTEM
    mem = ProjectMemory(db_path=tmp_path / "m.db")
    agent = MediaProducerAgent(_make_config(), memory=mem)
    task = Task(project="yolo", type="media", title="演示图",
                context="", priority=3, mode="auto", agent_role="media-producer")
    ctx = agent._build_context(task)
    assert len(ctx) > 10

Step 2: 运行确认失败

pytest tests/test_market_agents.py -v 2>&1 | head -15

Step 3: 创建 market_pm.py

from __future__ import annotations
from rockchip_agents.agents.developer import DeveloperAgent
from rockchip_agents.core.queue import Task

MARKET_PM_SYSTEM = """你是产品洞察与市场落地组的市场 PM，负责面向客户的交付文档。

输出类型：
- 方案书：技术方案 + 架构图 + 成本估算，保存至 docs/market/{project}-proposal.md
- 技术白皮书：平台能力 + 算法性能 + 对比竞品，保存至 docs/market/{project}-whitepaper.md
- BOM 估算：核心器件 + 单价区间 + 替代方案，保存至 docs/market/{project}-bom.md

内容要求：
- 基于 ProjectMemory 中的能力事实（FPS、精度、平台规格）填写实际数值
- 避免夸大性描述，数值来自测试报告
- 格式：Markdown，结构清晰，可直接发给客户

工作流：
1. 读取项目 docs/ 目录了解现有文档
2. 读取 ProjectMemory 获取能力数值
3. 生成或更新对应文档
"""


class MarketPmAgent(DeveloperAgent):
    def _build_context(self, task: Task) -> str:
        return super()._build_context(task) + "\n" + MARKET_PM_SYSTEM

Step 4: 创建 media_producer.py

from __future__ import annotations
from rockchip_agents.agents.developer import DeveloperAgent
from rockchip_agents.core.queue import Task

MEDIA_PRODUCER_SYSTEM = """你是产品洞察与市场落地组的媒体制作师，随时可被触发。

技术演示侧：
- 使用 OpenCV（python3 + cv2）在推理结果图上绘制检测框、置信度、平台水印
- 命令：python3 -c "import cv2; ..."，输出至 assets/demo/

市场展示侧：
- 生成产品效果图：硬件外观图 + 算法效果标注（Pillow 合成）
- Demo 视频剪辑：ffmpeg -i input.mp4 -vf "drawtext=..." output.mp4
- 输出至 assets/market/

工作流：
1. 确认输入素材路径（推理图/视频）
2. 选择合适的处理方式（cv2/ffmpeg/Pillow）
3. 执行并保存输出文件
4. 报告生成的文件路径

注意：可随时触发，不依赖测试结果，兼顾技术和市场两种用途。
"""


class MediaProducerAgent(DeveloperAgent):
    def _build_context(self, task: Task) -> str:
        return super()._build_context(task) + "\n" + MEDIA_PRODUCER_SYSTEM

Step 5: 运行测试

pytest tests/test_market_agents.py -v

---

Task 9: Executor 路由 — 注册所有新角色

Files:

• Modify: src/rockchip_agents/core/executor.py:42-63
• Test: tests/test_executor.py

Step 1: 写失败测试

在 tests/test_executor.py 末尾追加：

@pytest.mark.parametrize("role,expected_class", [
    ("os-engineer",    "OsEngineerAgent"),
    ("algo-researcher","AlgoResearcherAgent"),
    ("base-validator", "BaseValidatorAgent"),
    ("base-architect", "BaseArchitectAgent"),
    ("system-tester",  "SystemTesterAgent"),
    ("vision-analyst", "VisionAnalystAgent"),
    ("market-pm",      "MarketPmAgent"),
    ("media-producer", "MediaProducerAgent"),
])
def test_executor_routes_new_roles(role, expected_class, mock_config, tmp_path):
    from rockchip_agents.core.executor import _make_agent
    from rockchip_agents.core.queue import Task
    from rockchip_agents.tools.project_memory import ProjectMemory

    task = Task(project="test", type="test", title="t",
                context="", priority=3, mode="auto", agent_role=role)
    mem = ProjectMemory(db_path=tmp_path / "m.db")
    agent = _make_agent(mock_config, task, mem)
    assert type(agent).__name__ == expected_class, \
        f"role={role} 路由到 {type(agent).__name__}，期望 {expected_class}"

Step 2: 运行确认失败

pytest tests/test_executor.py::test_executor_routes_new_roles -v

Step 3: 修改 executor.py 的 _make_agent 函数

在现有 if role == "planner": 块之后，追加所有新角色：

if role == "os-engineer":
    from rockchip_agents.agents.os_engineer import OsEngineerAgent
    return OsEngineerAgent(config, memory=memory)
if role == "algo-researcher":
    from rockchip_agents.agents.algo_researcher import AlgoResearcherAgent
    return AlgoResearcherAgent(config, memory=memory)
if role == "base-validator":
    from rockchip_agents.agents.base_validator import BaseValidatorAgent
    return BaseValidatorAgent(config, memory=memory)
if role == "base-architect":
    from rockchip_agents.agents.base_architect import BaseArchitectAgent
    return BaseArchitectAgent(config, memory=memory)
if role == "system-tester":
    from rockchip_agents.agents.system_tester import SystemTesterAgent
    return SystemTesterAgent(config, memory=memory)
if role == "vision-analyst":
    from rockchip_agents.agents.vision_analyst import VisionAnalystAgent
    from rockchip_agents.core.queue import TaskQueue as _TQ
    return VisionAnalystAgent(config, queue=_TQ(), memory=memory)
if role == "market-pm":
    from rockchip_agents.agents.market_pm import MarketPmAgent
    return MarketPmAgent(config, memory=memory)
if role == "media-producer":
    from rockchip_agents.agents.media_producer import MediaProducerAgent
    return MediaProducerAgent(config, memory=memory)

Step 4: 全量回归

pytest tests/ -q 2>&1 | tail -10

预期：新增测试全部通过，已知 11 个集成失败不变。

---

验收标准

1. pytest tests/ -q 通过数 ≥ 160（新增 15+ 测试）
2. python -c "from rockchip_agents.config import load_config; cfg=load_config(); print(cfg.queues)" 输出三个队列配置
3. python -c "from rockchip_agents.core.executor import _make_agent" 可导入，15 个角色全部可路由
4. sqlite3 data/base_opt_tasks.db ".tables" 在 Scheduler 启动后自动创建