auto-sync: 2026-05-17 05:59:56

src/daemon/spawner.py

@@ -0,0 +1,218 @@
+"""Agent Spawner — 异步 spawn Full Agent / Subagent
+
+Full Agent: asyncio.create_subprocess_exec（异步非阻塞，不 await 完成）
+Subagent: 占位（实际通过 OpenClaw Gateway API sessions_spawn，F17 完善）
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import uuid
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Dict, Optional
+
+from src.blackboard.db import get_connection, init_db
+
+logger = logging.getLogger("moziplus-v2.spawner")
+
+
+class AgentSpawner:
+    """Agent spawn 管理"""
+
+    def __init__(
+        self,
+        db_path: Optional[Path] = None,
+        agent_timeout: float = 600.0,
+        dry_run: bool = False,
+    ):
+        """
+        Args:
+            db_path: 项目黑板 DB 路径（用于写 task_attempts）
+            agent_timeout: Agent 超时秒数
+            dry_run: 测试模式，不实际 spawn
+        """
+        self.db_path = db_path
+        self.agent_timeout = agent_timeout
+        self.dry_run = dry_run
+
+        # session 注册表 {session_id: {...}}
+        self._sessions: Dict[str, Dict[str, Any]] = {}
+
+    @property
+    def active_sessions(self) -> Dict[str, Dict[str, Any]]:
+        """当前活跃的 spawn sessions"""
+        return {sid: s for sid, s in self._sessions.items()
+                if s.get("status") == "running"}
+
+    async def spawn_full_agent(
+        self,
+        agent_id: str,
+        message: str,
+        new_session: bool = False,
+        task_id: Optional[str] = None,
+    ) -> str:
+        """Spawn Full Agent（异步非阻塞）
+
+        Returns:
+            session_id
+        """
+        session_id = str(uuid.uuid4())
+
+        if self.dry_run:
+            logger.info("[DRY RUN] Would spawn agent %s (session=%s)", agent_id, session_id)
+            self._register_session(session_id, agent_id, task_id, pid=None)
+            return session_id
+
+        cmd = [
+            "openclaw", "agent",
+            "--agent", agent_id,
+            "--session-id", session_id,
+            "--message", message,
+            "--json",
+        ]
+
+        try:
+            proc = await asyncio.create_subprocess_exec(
+                *cmd,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+            )
+            self._register_session(session_id, agent_id, task_id, proc.pid)
+            logger.info("Spawned agent %s (session=%s, pid=%d)",
+                        agent_id, session_id, proc.pid)
+
+            # Schedule timeout + cleanup
+            asyncio.create_task(
+                self._monitor_process(session_id, proc, agent_id, task_id)
+            )
+
+            return session_id
+
+        except Exception as e:
+            logger.exception("Failed to spawn agent %s", agent_id)
+            self._record_attempt(task_id, agent_id, "spawn_failed", error=str(e))
+            raise
+
+    async def spawn_subagent(
+        self,
+        task_description: str,
+        task_id: Optional[str] = None,
+    ) -> str:
+        """Spawn Subagent（占位，实际通过 Gateway API）
+
+        Returns:
+            session_id
+        """
+        session_id = str(uuid.uuid4())
+
+        if self.dry_run:
+            logger.info("[DRY RUN] Would spawn subagent (session=%s)", session_id)
+            self._register_session(session_id, "subagent", task_id, pid=None)
+            return session_id
+
+        # TODO: F17 通过 Gateway API sessions_spawn 实现
+        logger.info("Subagent spawn (session=%s) - placeholder", session_id)
+        self._register_session(session_id, "subagent", task_id, pid=None)
+        return session_id
+
+    async def _monitor_process(
+        self,
+        session_id: str,
+        proc: asyncio.subprocess.Process,
+        agent_id: str,
+        task_id: Optional[str],
+    ) -> None:
+        """监控子进程，超时 kill，完成后记录"""
+        try:
+            await asyncio.wait_for(proc.wait(), timeout=self.agent_timeout)
+            outcome = "completed"
+            exit_code = proc.returncode
+        except asyncio.TimeoutError:
+            proc.kill()
+            await proc.wait()
+            outcome = "timed_out"
+            exit_code = -1
+            logger.warning("Agent %s timed out (session=%s)", agent_id, session_id)
+
+        # 更新 session 状态
+        if session_id in self._sessions:
+            self._sessions[session_id]["status"] = outcome
+            self._sessions[session_id]["completed_at"] = datetime.utcnow().isoformat()
+
+        # 记录 task_attempt
+        self._record_attempt(task_id, agent_id, outcome, exit_code=exit_code)
+
+        logger.info("Agent %s finished (session=%s, outcome=%s, exit=%d)",
+                     agent_id, session_id, outcome, exit_code)
+
+    def _register_session(
+        self,
+        session_id: str,
+        agent_id: str,
+        task_id: Optional[str],
+        pid: Optional[int],
+    ) -> None:
+        """注册 spawn session"""
+        self._sessions[session_id] = {
+            "agent_id": agent_id,
+            "task_id": task_id,
+            "pid": pid,
+            "status": "running",
+            "started_at": datetime.utcnow().isoformat(),
+            "completed_at": None,
+        }
+
+    def _record_attempt(
+        self,
+        task_id: Optional[str],
+        agent_id: str,
+        outcome: str,
+        exit_code: Optional[int] = None,
+        error: Optional[str] = None,
+    ) -> None:
+        """记录 task_attempt"""
+        if not task_id or not self.db_path:
+            return
+
+        try:
+            conn = get_connection(self.db_path)
+            try:
+                conn.execute("BEGIN IMMEDIATE")
+                # 获取 attempt_number
+                row = conn.execute(
+                    "SELECT MAX(attempt_number) as max_a FROM task_attempts WHERE task_id=?",
+                    (task_id,),
+                ).fetchone()
+                attempt_number = (row["max_a"] or 0) + 1
+
+                metadata = {"error": error} if error else {}
+                conn.execute(
+                    "INSERT INTO task_attempts "
+                    "(task_id, attempt_number, agent, outcome, exit_code, metadata, completed_at) "
+                    "VALUES (?,?,?,?,?,?,datetime('now'))",
+                    (task_id, attempt_number, agent_id, outcome,
+                     exit_code, json.dumps(metadata)),
+                )
+                conn.execute(
+                    "INSERT INTO events (task_id, agent, event_type, detail) VALUES (?,?,?,?)",
+                    (task_id, agent_id,
+                     "agent_completed" if outcome == "completed" else "daemon_tick",
+                     json.dumps({"outcome": outcome, "attempt": attempt_number})),
+                )
+                conn.commit()
+            finally:
+                conn.close()
+        except Exception:
+            logger.exception("Failed to record attempt for task %s", task_id)
+
+    def get_session(self, session_id: str) -> Optional[Dict[str, Any]]:
+        """获取 session 信息"""
+        return self._sessions.get(session_id)
+
+    def cleanup_session(self, session_id: str) -> None:
+        """清理 session"""
+        if session_id in self._sessions:
+            del self._sessions[session_id]