cfdaily
965 lines
40 KiB
Python
965 lines
40 KiB
Python
"""Daemon Ticker — 30s 状态扫描 + 依赖推进 + 事件驱动
|
||
|
||
Tick 循环是整个 Daemon 的核心驱动：
|
||
1. 遍历所有 active 项目
|
||
2. 每个 tick 扫描黑板状态
|
||
3. 推进依赖链（blocked → pending）
|
||
4. 写入 daemon_tick 事件
|
||
5. 调度 pending 任务（F9 实现）
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Any, Callable, Coroutine, Dict, List, Optional
|
||
|
||
from src.blackboard.operations import Blackboard
|
||
from src.blackboard.db import get_connection
|
||
from src.blackboard.models import Task
|
||
from src.daemon.spawner import AgentBusyError
|
||
from src.blackboard.queries import Queries
|
||
from src.blackboard.registry import ProjectRegistry
|
||
|
||
logger = logging.getLogger("moziplus-v2.ticker")
|
||
|
||
|
||
class Ticker:
|
||
"""Daemon ticker 主循环"""
|
||
|
||
def __init__(
|
||
self,
|
||
registry: ProjectRegistry,
|
||
tick_interval: float = 30.0,
|
||
max_ticks: Optional[int] = None,
|
||
on_tick_complete: Optional[Callable[[], Coroutine[Any, Any, None]]] = None,
|
||
dispatcher: Optional[Any] = None,
|
||
spawner: Optional[Any] = None,
|
||
max_dispatch_per_tick: int = 3,
|
||
claim_timeout_minutes: float = 5.0,
|
||
default_task_timeout_minutes: float = 30.0,
|
||
health_checker: Optional[Any] = None,
|
||
experience_distiller: Optional[Any] = None,
|
||
inbox_watcher: Optional[Any] = None,
|
||
):
|
||
"""
|
||
Args:
|
||
registry: 项目注册表
|
||
tick_interval: tick 间隔秒数
|
||
max_ticks: 测试用，限制最大 tick 次数
|
||
on_tick_complete: 每次 tick 完成后的回调（用于通知等）
|
||
dispatcher: Dispatcher 实例（Agent 调度）
|
||
spawner: AgentSpawner 实例（Agent spawn）
|
||
max_dispatch_per_tick: 每个 tick 最多调度多少个 pending 任务
|
||
claim_timeout_minutes: claimed 超时重置为 pending 的分钟数
|
||
default_task_timeout_minutes: working 超时标记 failed 的分钟数
|
||
"""
|
||
self.registry = registry
|
||
self.tick_interval = tick_interval
|
||
self.max_ticks = max_ticks
|
||
self.on_tick_complete = on_tick_complete
|
||
self.dispatcher = dispatcher
|
||
self.spawner = spawner
|
||
self.max_dispatch_per_tick = max_dispatch_per_tick
|
||
self.claim_timeout_minutes = claim_timeout_minutes
|
||
self.default_task_timeout_minutes = default_task_timeout_minutes
|
||
self.health_checker = health_checker
|
||
self.experience_distiller = experience_distiller
|
||
self.inbox_watcher = inbox_watcher
|
||
|
||
self._tick_count: int = 0
|
||
self._running: bool = False
|
||
self._task: Optional[asyncio.Task] = None
|
||
|
||
# 已初始化的 db_path 集合，避免每次 tick 重复 init_db
|
||
self._initialized_dbs: set = set()
|
||
|
||
# 每个项目上次 tick 的 event count（用于僵尸检测 F8）
|
||
self._last_event_counts: Dict[str, int] = {}
|
||
|
||
# 当前项目 ID（_dispatch_pending 需要）
|
||
self._current_project_id: Optional[str] = None
|
||
|
||
# ------------------------------------------------------------------
|
||
# 生命周期
|
||
# ------------------------------------------------------------------
|
||
|
||
async def start(self) -> None:
|
||
"""启动 tick 循环"""
|
||
if self._running:
|
||
return
|
||
self._running = True
|
||
self._task = asyncio.create_task(self._loop())
|
||
# 启动 InboxWatcher（即时事件监听）
|
||
if self.inbox_watcher:
|
||
try:
|
||
await self.inbox_watcher.start()
|
||
logger.info("InboxWatcher started")
|
||
except Exception as e:
|
||
logger.warning("InboxWatcher start failed: %s", e)
|
||
logger.info("Ticker started (interval=%.1fs)", self.tick_interval)
|
||
|
||
async def stop(self) -> None:
|
||
"""停止 tick 循环"""
|
||
self._running = False
|
||
# 停止 InboxWatcher
|
||
if self.inbox_watcher:
|
||
try:
|
||
await self.inbox_watcher.stop()
|
||
logger.info("InboxWatcher stopped")
|
||
except Exception as e:
|
||
logger.warning("InboxWatcher stop failed: %s", e)
|
||
if self._task:
|
||
self._task.cancel()
|
||
try:
|
||
await self._task
|
||
except asyncio.CancelledError:
|
||
pass
|
||
logger.info("Ticker stopped (ticks=%d)", self._tick_count)
|
||
|
||
@property
|
||
def tick_count(self) -> int:
|
||
return self._tick_count
|
||
|
||
@property
|
||
def is_running(self) -> bool:
|
||
return self._running
|
||
|
||
# ------------------------------------------------------------------
|
||
# 主循环
|
||
# ------------------------------------------------------------------
|
||
|
||
async def _loop(self) -> None:
|
||
"""Tick 主循环"""
|
||
while self._running:
|
||
try:
|
||
await self.tick()
|
||
except asyncio.CancelledError:
|
||
raise
|
||
except Exception:
|
||
logger.exception("Tick %d error", self._tick_count + 1)
|
||
|
||
if self.max_ticks and self._tick_count >= self.max_ticks:
|
||
logger.info("Max ticks reached (%d), stopping", self.max_ticks)
|
||
self._running = False
|
||
break
|
||
|
||
try:
|
||
await asyncio.sleep(self.tick_interval)
|
||
except asyncio.CancelledError:
|
||
return
|
||
|
||
# ------------------------------------------------------------------
|
||
# 单次 tick
|
||
# ------------------------------------------------------------------
|
||
|
||
async def tick(self) -> Dict[str, Any]:
|
||
"""执行一次 tick，遍历所有 active 项目"""
|
||
self._tick_count += 1
|
||
tick_num = self._tick_count
|
||
results: Dict[str, Any] = {
|
||
"tick": tick_num,
|
||
"projects": {},
|
||
}
|
||
|
||
projects = self.registry.list_projects()
|
||
active_projects = {
|
||
pid: info for pid, info in projects.items()
|
||
if info.get("status") == "active"
|
||
}
|
||
|
||
for project_id, project_info in active_projects.items():
|
||
try:
|
||
pr = await self._tick_project(project_id, project_info)
|
||
results["projects"][project_id] = pr
|
||
except Exception as e:
|
||
logger.exception("Tick %d project %s error", tick_num, project_id)
|
||
results["projects"][project_id] = {"error": str(e)}
|
||
|
||
# 虚拟项目 _general：不在 registry 但需要调度
|
||
general_db = Path(self.registry.root) / "_general" / "blackboard.db"
|
||
if general_db.exists() and "_general" not in active_projects:
|
||
try:
|
||
pr = await self._tick_project("_general", {
|
||
"id": "_general", "name": "一般任务",
|
||
"status": "active", "source": "virtual",
|
||
})
|
||
results["projects"]["_general"] = pr
|
||
except Exception as e:
|
||
logger.exception("Tick %d _general error", tick_num)
|
||
results["projects"]["_general"] = {"error": str(e)}
|
||
|
||
# 虚拟项目 _mail：飞鸽传书
|
||
mail_db = Path(self.registry.root) / "_mail" / "blackboard.db"
|
||
if mail_db.exists() and "_mail" not in active_projects:
|
||
try:
|
||
pr = await self._tick_project("_mail", {
|
||
"id": "_mail", "name": "飞鸽传书",
|
||
"status": "active", "source": "virtual",
|
||
})
|
||
results["projects"]["_mail"] = pr
|
||
except Exception as e:
|
||
logger.exception("Tick %d _mail error", tick_num)
|
||
results["projects"]["_mail"] = {"error": str(e)}
|
||
|
||
logger.debug("Tick %d complete: %d projects", tick_num, len(active_projects))
|
||
|
||
if self.on_tick_complete:
|
||
try:
|
||
await self.on_tick_complete()
|
||
except Exception:
|
||
logger.exception("on_tick_complete callback error")
|
||
|
||
return results
|
||
|
||
async def _tick_project(self, project_id: str,
|
||
project_info: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""单项目的 tick 处理"""
|
||
project_dir = self.registry.root / project_id
|
||
db_path = project_dir / "blackboard.db"
|
||
|
||
if not db_path.exists():
|
||
return {"status": "no_db"}
|
||
|
||
# 只在首次遇到该 db_path 时执行 init_db
|
||
db_key = str(db_path)
|
||
if db_key not in self._initialized_dbs:
|
||
from src.blackboard.db import init_db
|
||
init_db(db_path)
|
||
self._initialized_dbs.add(db_key)
|
||
|
||
result: Dict[str, Any] = {
|
||
"status": "ok",
|
||
"summary_before": {},
|
||
"summary_after": {},
|
||
"advanced": [],
|
||
"dispatched": [],
|
||
"review_dispatched": [],
|
||
"zombie_reclaimed": [],
|
||
"parents_refreshed": [],
|
||
}
|
||
|
||
# 保存当前 project_id（供 _dispatch_pending 使用）
|
||
self._current_project_id = project_id
|
||
|
||
# 1. 扫描当前状态
|
||
queries = Queries(db_path)
|
||
result["summary_before"] = queries.task_summary()
|
||
|
||
# 2. 依赖推进
|
||
advanced = self._advance_dependencies(db_path)
|
||
result["advanced"] = advanced
|
||
|
||
# 3. 僵尸/超时处理（在依赖推进后、调度前）
|
||
zombie_reclaimed = self._check_timeouts(db_path)
|
||
result["zombie_reclaimed"] = zombie_reclaimed
|
||
|
||
# 4. 调度 pending 任务
|
||
if self.dispatcher and self.spawner:
|
||
dispatched = await self._dispatch_pending(db_path, project_id)
|
||
result["dispatched"] = dispatched
|
||
|
||
# 5. 调度审查任务
|
||
if self.dispatcher and self.spawner:
|
||
review_dispatched = await self._dispatch_reviews(db_path, project_id)
|
||
result["review_dispatched"] = review_dispatched
|
||
|
||
# 6. 聚合父 Task 状态（v2.7）
|
||
parents_refreshed = self._refresh_parent_statuses(db_path)
|
||
result["parents_refreshed"] = parents_refreshed
|
||
|
||
# 7. 一轮结束检测 + 庞统 review spawn（v2.9 #01）
|
||
round_reviewed = await self._check_round_complete(db_path, project_id)
|
||
result["round_reviewed"] = round_reviewed
|
||
|
||
# 7. 写 daemon_tick 事件
|
||
conn = get_connection(db_path)
|
||
try:
|
||
conn.execute("BEGIN IMMEDIATE")
|
||
conn.execute(
|
||
"INSERT INTO events (task_id, agent, event_type, detail) VALUES (?,?,?,?)",
|
||
(None, "daemon", "daemon_tick",
|
||
json.dumps({"tick": self._tick_count, "advanced_count": len(advanced),
|
||
"parents_refreshed": len(parents_refreshed)})),
|
||
)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
# 8. 健康检查（僵尸检测）
|
||
if self.health_checker:
|
||
try:
|
||
self.health_checker.check(project_id, db_path, self._tick_count)
|
||
except Exception as e:
|
||
logger.warning("HealthChecker error for %s: %s", project_id, e)
|
||
|
||
# 9. 经验蒸馏（完成的 task 自动触发）
|
||
if self.experience_distiller:
|
||
try:
|
||
conn2 = get_connection(db_path)
|
||
try:
|
||
done_tasks = conn2.execute(
|
||
"SELECT id FROM tasks WHERE status='done' AND updated_at > datetime('now', '-60 seconds')"
|
||
).fetchall()
|
||
finally:
|
||
conn2.close()
|
||
for row in done_tasks:
|
||
t = Blackboard(db_path).get_task(row[0])
|
||
if t:
|
||
self.experience_distiller.distill_from_task(
|
||
task_id=t.id, task_title=t.title, task_type=t.task_type
|
||
)
|
||
except Exception as e:
|
||
logger.warning("ExperienceDistiller error for %s: %s", project_id, e)
|
||
|
||
# 10. 扫描后状态
|
||
result["summary_after"] = queries.task_summary()
|
||
|
||
return result
|
||
|
||
# ------------------------------------------------------------------
|
||
# 父 Task 状态聚合
|
||
# ------------------------------------------------------------------
|
||
|
||
def _refresh_parent_statuses(self, db_path: Path) -> List[str]:
|
||
"""全量扫描有子 Task 的父 Task，刷新聚合状态
|
||
|
||
跳过手动状态（cancelled, paused）的父 Task。使用单连接批量处理。
|
||
"""
|
||
queries = Queries(db_path)
|
||
refreshed: List[str] = []
|
||
|
||
conn = get_connection(db_path)
|
||
try:
|
||
# 找出所有有子 Task 的父 Task ID
|
||
parent_rows = conn.execute(
|
||
"SELECT DISTINCT parent_task FROM tasks WHERE parent_task IS NOT NULL"
|
||
).fetchall()
|
||
parent_ids = [r["parent_task"] for r in parent_rows]
|
||
|
||
for pid in parent_ids:
|
||
computed = queries.compute_parent_status(pid)
|
||
if computed is None:
|
||
continue
|
||
parent = conn.execute(
|
||
"SELECT status FROM tasks WHERE id=?", (pid,)
|
||
).fetchone()
|
||
if parent and parent["status"] != computed:
|
||
conn.execute(
|
||
"UPDATE tasks SET status=?, updated_at=datetime('now') WHERE id=?",
|
||
(computed, pid),
|
||
)
|
||
refreshed.append(pid)
|
||
logger.info("Parent %s status aggregated: → %s", pid, computed)
|
||
|
||
if refreshed:
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
return refreshed
|
||
|
||
# ------------------------------------------------------------------
|
||
# 依赖推进
|
||
# ------------------------------------------------------------------
|
||
|
||
def _advance_dependencies(self, db_path: Path) -> List[str]:
|
||
"""检查 blocked 任务，若所有依赖已完成则推进为 pending
|
||
|
||
Returns:
|
||
被推进的 task_id 列表
|
||
"""
|
||
queries = Queries(db_path)
|
||
blocked = queries.blocked_tasks_with_deps()
|
||
advanced: List[str] = []
|
||
|
||
if not blocked:
|
||
return advanced
|
||
|
||
conn = get_connection(db_path)
|
||
try:
|
||
for item in blocked:
|
||
if item["all_deps_done"]:
|
||
task_id = item["task_id"]
|
||
ok = self._transition_status(conn, task_id, "pending",
|
||
agent="daemon",
|
||
detail={"reason": "all_dependencies_done"})
|
||
if ok:
|
||
advanced.append(task_id)
|
||
logger.info("Advanced %s: blocked → pending", task_id)
|
||
finally:
|
||
conn.close()
|
||
|
||
return advanced
|
||
|
||
def _transition_status(self, conn, task_id: str, new_status: str,
|
||
agent: str = "daemon",
|
||
detail: Optional[Dict] = None) -> bool:
|
||
"""轻量状态转换（不走 Blackboard 类，避免 init_db）"""
|
||
from src.blackboard.db import VALID_TRANSITIONS, TERMINAL_STATUSES, EVENT_TYPES
|
||
from datetime import datetime
|
||
|
||
conn.execute("BEGIN IMMEDIATE")
|
||
row = conn.execute("SELECT status FROM tasks WHERE id=?", (task_id,)).fetchone()
|
||
if not row:
|
||
return False
|
||
old_status = row["status"]
|
||
if old_status in TERMINAL_STATUSES or old_status == new_status:
|
||
return old_status == new_status
|
||
if new_status not in VALID_TRANSITIONS.get(old_status, set()):
|
||
return False
|
||
|
||
now = datetime.utcnow().isoformat()
|
||
# 重置到 pending 时清空 assignee（避免残留导致重复路由到同一 Agent）
|
||
# 但 Mail 的 assignee 是收件人，永不清空
|
||
if new_status == "pending":
|
||
if self._current_project_id == "_mail":
|
||
conn.execute(
|
||
"UPDATE tasks SET status=?, updated_at=? WHERE id=?",
|
||
(new_status, now, task_id),
|
||
)
|
||
else:
|
||
conn.execute(
|
||
"UPDATE tasks SET status=?, assignee=NULL, resumed_from=NULL, updated_at=? WHERE id=?",
|
||
(new_status, now, task_id),
|
||
)
|
||
elif new_status == "paused":
|
||
# 记录暂停前状态，恢复时回到原状态
|
||
conn.execute(
|
||
"UPDATE tasks SET status=?, resumed_from=?, updated_at=? WHERE id=?",
|
||
(new_status, old_status, now, task_id),
|
||
)
|
||
else:
|
||
conn.execute(
|
||
"UPDATE tasks SET status=?, updated_at=? WHERE id=?",
|
||
(new_status, now, task_id),
|
||
)
|
||
event_type = f"task_{new_status}"
|
||
if event_type not in EVENT_TYPES:
|
||
event_type = "daemon_tick"
|
||
conn.execute(
|
||
"INSERT INTO events (task_id, agent, event_type, detail) VALUES (?,?,?,?)",
|
||
(task_id, agent, event_type, json.dumps({"from": old_status, "to": new_status, **(detail or {})})),
|
||
)
|
||
conn.commit()
|
||
return True
|
||
|
||
# ------------------------------------------------------------------
|
||
# Agent 调度
|
||
# ------------------------------------------------------------------
|
||
|
||
async def _dispatch_pending(self, db_path: Path,
|
||
project_id: str) -> List[str]:
|
||
"""扫描 pending 任务并调度
|
||
|
||
v3.0: 两条路径
|
||
- 确定性路径：retry/handoff/assignee/生命周期 → 直接 spawn
|
||
- 广播认领：无确定性路径 → spawn 所有空闲 Agent，自主 claim
|
||
"""
|
||
queries = Queries(db_path)
|
||
pending = queries.pending_dispatchable()
|
||
dispatched: List[str] = []
|
||
|
||
if not pending:
|
||
return dispatched
|
||
|
||
# 分两批：确定性 vs 广播
|
||
deterministic_tasks = []
|
||
broadcast_tasks = []
|
||
|
||
for task in pending[:self.max_dispatch_per_tick]:
|
||
decision = self.dispatcher.decide(task)
|
||
if decision.get("mode") in ("deterministic", "agent_handoff"):
|
||
deterministic_tasks.append((task, decision))
|
||
else:
|
||
broadcast_tasks.append(task)
|
||
|
||
# 路径1：确定性路由 → 直接 spawn
|
||
for task, decision in deterministic_tasks:
|
||
try:
|
||
result = await self.dispatcher.dispatch(
|
||
task,
|
||
project_config={"project_id": project_id, "db_path": db_path},
|
||
)
|
||
if result["status"] == "dispatched" and result["level"] in ("full", "escalate"):
|
||
conn = get_connection(db_path)
|
||
try:
|
||
# [v2.7.1] Mail 已在 dispatcher 中标 working，跳过 claimed
|
||
if project_id == "_mail":
|
||
conn.execute(
|
||
"UPDATE tasks SET current_agent=? WHERE id=?",
|
||
(result["agent_id"], task.id),
|
||
)
|
||
conn.commit()
|
||
dispatched.append(task.id)
|
||
logger.info("Dispatched %s to %s (session=%s, mail auto-working)",
|
||
task.id, result["agent_id"],
|
||
result.get("session_id"))
|
||
else:
|
||
ok = self._transition_status(
|
||
conn, task.id, "claimed",
|
||
agent="daemon",
|
||
detail={"dispatched_to": result["agent_id"],
|
||
"session_id": result.get("session_id")},
|
||
)
|
||
if ok:
|
||
conn.execute(
|
||
"UPDATE tasks SET current_agent=? WHERE id=?",
|
||
(result["agent_id"], task.id),
|
||
)
|
||
conn.commit()
|
||
dispatched.append(task.id)
|
||
logger.info("Dispatched %s to %s (session=%s)",
|
||
task.id, result["agent_id"],
|
||
result.get("session_id"))
|
||
finally:
|
||
conn.close()
|
||
elif result["status"] == "blocked":
|
||
# Guardrail 拦截：任务标记为 blocked
|
||
conn = get_connection(db_path)
|
||
try:
|
||
self._transition_status(
|
||
conn, task.id, "blocked",
|
||
agent="daemon",
|
||
detail={"reason": result.get("reason", "guardrail"),
|
||
"violations": result.get("violations", [])},
|
||
)
|
||
logger.info("Task %s blocked by guardrail: %s",
|
||
task.id, result.get("reason", "unknown"))
|
||
finally:
|
||
conn.close()
|
||
except Exception:
|
||
logger.exception("Dispatch failed for %s", task.id)
|
||
|
||
# 路径2：广播认领
|
||
if broadcast_tasks:
|
||
broadcast_ids = await self._broadcast_claim(broadcast_tasks, db_path, project_id)
|
||
dispatched.extend(broadcast_ids)
|
||
|
||
return dispatched
|
||
|
||
async def _broadcast_claim(self, tasks: list, db_path: Path,
|
||
project_id: str) -> List[str]:
|
||
"""广播一批待认领任务给所有空闲 Agent（每轮最多广播一次）
|
||
|
||
司马懿建议：攒一批任务，一次广播，而非每个任务触发一次广播。
|
||
广播前检查全局并发，接近上限时跳过。
|
||
"""
|
||
if not self.spawner:
|
||
return []
|
||
|
||
# 全局并发检查（司马懿建议 1）
|
||
if self.counter.is_near_limit():
|
||
logger.info("Skipping broadcast: global concurrent near limit (%d/%d)",
|
||
self.counter.global_active, self.counter.max_global)
|
||
return []
|
||
|
||
# 过滤掉已广播太多次的任务（retry_count >= 3 → 不广播，等庞统）
|
||
broadcastable = []
|
||
escalated = []
|
||
for t in tasks:
|
||
rc = getattr(t, 'retry_count', 0) or 0
|
||
if rc >= 3:
|
||
escalated.append(t)
|
||
else:
|
||
broadcastable.append(t)
|
||
|
||
# 升级庞统
|
||
for t in escalated:
|
||
conn = get_connection(db_path)
|
||
try:
|
||
self._transition_status(
|
||
conn, t.id, "escalated",
|
||
agent="daemon",
|
||
detail={"reason": "no_taker_after_3_broadcasts",
|
||
"retry_count": getattr(t, 'retry_count', 0)},
|
||
)
|
||
logger.warning("Escalated %s: no taker after %d broadcasts",
|
||
t.id, getattr(t, 'retry_count', 0))
|
||
finally:
|
||
conn.close()
|
||
|
||
if not broadcastable:
|
||
return []
|
||
|
||
# 获取空闲 Agent
|
||
idle_agents = self._get_idle_agents()
|
||
if not idle_agents:
|
||
# 无空闲 Agent → 系统容量问题，不递增 retry_count
|
||
logger.warning("No idle agents for broadcast, skipping (capacity issue)")
|
||
return []
|
||
|
||
task_ids = [t.id for t in broadcastable]
|
||
logger.info("Broadcasting %d tasks to %d idle agents: %s",
|
||
len(broadcastable), len(idle_agents), task_ids)
|
||
|
||
# 广播前为每个任务写审计事件
|
||
try:
|
||
conn = get_connection(db_path)
|
||
try:
|
||
for task in broadcastable:
|
||
conn.execute(
|
||
"INSERT INTO events (task_id, agent, event_type, detail) "
|
||
"VALUES (?,?,?,?)",
|
||
(task.id, "daemon", "broadcast_claim",
|
||
json.dumps({"agents": idle_agents, "task_title": task.title})),
|
||
)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
except Exception:
|
||
pass # 审计记录失败不影响广播
|
||
|
||
spawned = []
|
||
for agent_id in idle_agents:
|
||
# v2.7.2: counter 检查在 spawn_full_agent 内部
|
||
prompt = self._build_claim_prompt(agent_id, broadcastable, project_id)
|
||
try:
|
||
session_id = await self.spawner.spawn_full_agent(
|
||
agent_id=agent_id,
|
||
message=prompt,
|
||
# v2.7.2: on_complete 不含 counter.release，由 wrapped_on_complete 保证
|
||
task_id=broadcastable[0].id if broadcastable else None,
|
||
task_db_path=db_path,
|
||
)
|
||
spawned.append(agent_id)
|
||
logger.info("Broadcast spawned %s (session=%s)", agent_id, session_id)
|
||
except AgentBusyError:
|
||
logger.debug("Broadcast skip %s: busy", agent_id)
|
||
except Exception:
|
||
logger.exception("Broadcast spawn failed for %s", agent_id)
|
||
|
||
return task_ids if spawned else []
|
||
|
||
def _build_claim_prompt(self, agent_id: str, tasks: list,
|
||
project_id: str) -> str:
|
||
"""构建广播认领的 prompt"""
|
||
api_host = getattr(self.spawner, 'api_host', '127.0.0.1') if self.spawner else '127.0.0.1'
|
||
api_port = getattr(self.spawner, 'api_port', 8083) if self.spawner else 8083
|
||
|
||
task_list = "\n".join([
|
||
f"- ID: {t.id}, 标题: {t.title}, 类型: {getattr(t, 'task_type', '') or 'general'}, "
|
||
f"优先级: {t.priority}"
|
||
for t in tasks
|
||
])
|
||
|
||
return f"""你是 {agent_id}。黑板上有 {len(tasks)} 个待认领任务。
|
||
|
||
## 重要规则
|
||
- 只认领符合你专长的任务。如果你不是最合适的执行者，不要认领
|
||
- 认领后必须写产出物再转 review，不能无产出就提交
|
||
|
||
## 待认领任务
|
||
{task_list}
|
||
|
||
## 操作
|
||
1. 读黑板查看详情：
|
||
curl http://{api_host}:{api_port}/api/projects/{project_id}/tasks?status=pending
|
||
|
||
2. 选择适合你专长的任务并认领：
|
||
curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/TASK_ID/claim \\
|
||
-H 'Content-Type: application/json' -d '{{"agent": "{agent_id}"}}'
|
||
|
||
3. 认领成功后开始执行：
|
||
curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/TASK_ID/status \\
|
||
-H 'Content-Type: application/json' -d '{{"status": "working", "agent": "{agent_id}"}}'
|
||
|
||
4. 没有适合你的任务则退出
|
||
"""
|
||
|
||
@property
|
||
def counter(self):
|
||
"""从 Dispatcher 获取 counter"""
|
||
return getattr(self.dispatcher, 'counter', None) if self.dispatcher else None
|
||
|
||
@staticmethod
|
||
def _is_pid_alive(pid: int) -> bool:
|
||
"""检查进程是否存活"""
|
||
import os
|
||
try:
|
||
os.kill(pid, 0)
|
||
return True
|
||
except (ProcessLookupError, PermissionError):
|
||
return False
|
||
|
||
def _get_idle_agents(self) -> List[str]:
|
||
"""获取当前空闲的 Agent 列表（从 config agents 取，而非 counter._per_agent）"""
|
||
if not self.counter:
|
||
return []
|
||
# agent_profiles 在 Router 初始化时从 config 填充，是完整 Agent 列表
|
||
all_agents = list(self.dispatcher.router.agent_profiles.keys()) if self.dispatcher else []
|
||
active = self.counter.active_agents
|
||
return [aid for aid in all_agents if active.get(aid, 0) == 0]
|
||
|
||
async def _dispatch_reviews(self, db_path: Path,
|
||
project_id: str) -> List[str]:
|
||
"""扫描 review 状态任务，检查是否有产出，调度审查 Agent"""
|
||
# mail 任务不走 review 流程，直接跳过
|
||
if project_id == "_mail":
|
||
return []
|
||
|
||
queries = Queries(db_path)
|
||
bb = Blackboard(db_path)
|
||
review_tasks = queries.tasks_by_status("review")
|
||
dispatched: List[str] = []
|
||
|
||
for task in review_tasks:
|
||
# 检查是否已有 review 记录
|
||
reviews = bb.get_reviews(task.id)
|
||
if reviews:
|
||
continue # 已有审查，跳过
|
||
|
||
# 检查是否最近已 dispatch 过 review（防重复 dispatch）
|
||
existing = self._check_recent_routing(db_path, task.id, "review")
|
||
if existing:
|
||
continue # 已有活跃 review dispatch
|
||
|
||
# 检查是否有产出（司马懿建议：无产出直接标 failed）
|
||
outputs = bb.get_outputs(task.id)
|
||
if not outputs:
|
||
conn = get_connection(db_path)
|
||
try:
|
||
self._transition_status(
|
||
conn, task.id, "failed",
|
||
agent="daemon",
|
||
detail={"reason": "no_output_for_review"},
|
||
)
|
||
bb.add_observation(
|
||
task.id, "daemon",
|
||
"任务进入 review 状态但没有产出物，自动标记为 failed",
|
||
)
|
||
logger.warning("Task %s in review but no output, marking failed",
|
||
task.id)
|
||
finally:
|
||
conn.close()
|
||
continue
|
||
|
||
# 调度审查 Agent（司马懿）
|
||
try:
|
||
result = await self.dispatcher.dispatch(
|
||
task,
|
||
action_type="review",
|
||
project_config={"project_id": project_id, "db_path": db_path},
|
||
)
|
||
if result["status"] == "dispatched":
|
||
dispatched.append(task.id)
|
||
# 更新 current_agent 为审查者
|
||
conn = get_connection(db_path)
|
||
try:
|
||
conn.execute(
|
||
"UPDATE tasks SET current_agent=? WHERE id=?",
|
||
(result["agent_id"], task.id),
|
||
)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
logger.info("Dispatched review for %s to %s",
|
||
task.id, result["agent_id"])
|
||
except Exception:
|
||
logger.exception("Review dispatch failed for %s", task.id)
|
||
|
||
return dispatched
|
||
|
||
# ------------------------------------------------------------------
|
||
# 僵尸/超时处理
|
||
# ------------------------------------------------------------------
|
||
|
||
def _check_timeouts(self, db_path: Path) -> List[str]:
|
||
"""检查 claimed/working 超时的任务"""
|
||
queries = Queries(db_path)
|
||
reclaimed: List[str] = []
|
||
now = datetime.utcnow() # UTC，与 SQLite datetime('now') 一致
|
||
|
||
# claimed 超时 → 重置为 pending（如果 retry_count >= 3 则升级庞统）
|
||
claimed = queries.tasks_by_status("claimed")
|
||
for task in claimed:
|
||
if not task.claimed_at:
|
||
continue
|
||
try:
|
||
claimed_time = datetime.fromisoformat(task.claimed_at)
|
||
elapsed = (now - claimed_time).total_seconds() / 60.0
|
||
if elapsed > self.claim_timeout_minutes:
|
||
retry_count = getattr(task, 'retry_count', 0) or 0
|
||
# 司马懿建议：复用 retry_count，>=3 则升级庞统
|
||
if retry_count >= 3:
|
||
conn = get_connection(db_path)
|
||
try:
|
||
self._transition_status(
|
||
conn, task.id, "escalated",
|
||
agent="daemon",
|
||
detail={"reason": "claim_timeout_no_taker",
|
||
"retry_count": retry_count},
|
||
)
|
||
reclaimed.append(task.id)
|
||
logger.warning("Escalated %s: no taker after %d broadcasts",
|
||
task.id, retry_count)
|
||
finally:
|
||
conn.close()
|
||
else:
|
||
conn = get_connection(db_path)
|
||
try:
|
||
ok = self._transition_status(
|
||
conn, task.id, "pending",
|
||
agent="daemon",
|
||
detail={"reason": "claim_timeout",
|
||
"elapsed_minutes": round(elapsed, 1)},
|
||
)
|
||
if ok:
|
||
# 递增 retry_count（复用为广播轮次计数）
|
||
conn.execute(
|
||
"UPDATE tasks SET retry_count = COALESCE(retry_count, 0) + 1 WHERE id=?",
|
||
(task.id,),
|
||
)
|
||
conn.commit()
|
||
reclaimed.append(task.id)
|
||
logger.info("Reclaimed %s: claimed → pending (timeout %.1fm, retry=%d)",
|
||
task.id, elapsed, retry_count + 1)
|
||
finally:
|
||
conn.close()
|
||
except (ValueError, TypeError):
|
||
pass
|
||
|
||
# working 超时 → 标记为 failed
|
||
working = queries.tasks_by_status("working")
|
||
for task in working:
|
||
start_time_str = task.started_at or task.claimed_at
|
||
if not start_time_str:
|
||
continue
|
||
try:
|
||
start_time = datetime.fromisoformat(start_time_str)
|
||
# per-task timeout: deadline 优先，否则用默认值
|
||
if task.deadline:
|
||
deadline_time = datetime.fromisoformat(task.deadline)
|
||
timeout_minutes = (deadline_time - start_time).total_seconds() / 60.0
|
||
if timeout_minutes < 1:
|
||
timeout_minutes = self.default_task_timeout_minutes
|
||
else:
|
||
timeout_minutes = self.default_task_timeout_minutes
|
||
|
||
elapsed = (now - start_time).total_seconds() / 60.0
|
||
if elapsed > timeout_minutes:
|
||
# [v2.7.1] Mail 幻觉门控兜底：有回复 + working → done
|
||
if self._current_project_id == "_mail":
|
||
has_reply = self._mail_check_reply(task.id, db_path)
|
||
if has_reply:
|
||
conn = get_connection(db_path)
|
||
try:
|
||
ok = self._transition_status(
|
||
conn, task.id, "done",
|
||
agent="daemon",
|
||
detail={"reason": "mail_auto_done_recheck",
|
||
"elapsed_minutes": round(elapsed, 1)},
|
||
)
|
||
if ok:
|
||
reclaimed.append(task.id)
|
||
logger.info("Mail %s: ticker recheck found reply, marked done (%.1fm)",
|
||
task.id, elapsed)
|
||
finally:
|
||
conn.close()
|
||
continue
|
||
|
||
conn = get_connection(db_path)
|
||
try:
|
||
ok = self._transition_status(
|
||
conn, task.id, "failed",
|
||
agent="daemon",
|
||
detail={"reason": "task_timeout",
|
||
"elapsed_minutes": round(elapsed, 1),
|
||
"timeout_minutes": round(timeout_minutes, 1)},
|
||
)
|
||
if ok:
|
||
reclaimed.append(task.id)
|
||
logger.warning("Task %s timed out (working %.1fm > %.1fm)",
|
||
task.id, elapsed, timeout_minutes)
|
||
finally:
|
||
conn.close()
|
||
except (ValueError, TypeError):
|
||
pass
|
||
|
||
# v2.7.2: 进程存活性检查 — counter 占用但进程已死的兜底
|
||
if self.spawner and self.counter:
|
||
for agent_id in list(self.counter.active_agents.keys()):
|
||
session_info = self.spawner.get_session_by_agent(agent_id)
|
||
if not session_info:
|
||
continue
|
||
pid = session_info.get("pid")
|
||
task_id_check = session_info.get("task_id")
|
||
if pid and not self._is_pid_alive(pid):
|
||
logger.warning("Agent %s process dead (pid=%d), releasing counter + push back pending",
|
||
agent_id, pid)
|
||
self.counter.release(agent_id)
|
||
# 推回 pending 让 ticker 重新 dispatch
|
||
if task_id_check and db_path:
|
||
try:
|
||
conn = get_connection(db_path)
|
||
try:
|
||
self._transition_status(
|
||
conn, task_id_check, "pending",
|
||
agent="daemon",
|
||
detail={"reason": "process_dead", "pid": pid},
|
||
)
|
||
finally:
|
||
conn.close()
|
||
except Exception:
|
||
logger.exception("Failed to push back task %s", task_id_check)
|
||
|
||
return reclaimed
|
||
|
||
def _mail_check_reply(self, original_task_id: str, db_path: Path) -> bool:
|
||
"""Mail 幻觉门控：检查是否有回复邮件"""
|
||
try:
|
||
conn = get_connection(db_path)
|
||
try:
|
||
row = conn.execute(
|
||
"SELECT id FROM tasks WHERE id != ? AND must_haves LIKE ? LIMIT 1",
|
||
(original_task_id, f'%{original_task_id}%'),
|
||
).fetchone()
|
||
return row is not None
|
||
finally:
|
||
conn.close()
|
||
except Exception as e:
|
||
logger.error("Mail %s: ticker reply check error: %s", original_task_id, e)
|
||
return True # 保守：查询失败假设有回复
|
||
|
||
def _check_recent_routing(self, db_path: Path, task_id: str,
|
||
action_type: str) -> bool:
|
||
"""检查最近 5 分钟内是否已 dispatch 过指定类型的路由（防重复）"""
|
||
try:
|
||
conn = get_connection(db_path)
|
||
try:
|
||
# 检查是否有 from_status=review 的 dispatched 记录（防止重复 review dispatch）
|
||
if action_type == "review":
|
||
row = conn.execute(
|
||
"SELECT COUNT(*) as cnt FROM routing_decisions "
|
||
"WHERE task_id=? AND outcome='dispatched' "
|
||
"AND from_status='review' "
|
||
"AND created_at > datetime('now', '-5 minutes')",
|
||
(task_id,),
|
||
).fetchone()
|
||
else:
|
||
row = conn.execute(
|
||
"SELECT COUNT(*) as cnt FROM routing_decisions "
|
||
"WHERE task_id=? AND outcome='dispatched' "
|
||
"AND created_at > datetime('now', '-5 minutes')",
|
||
(task_id,),
|
||
).fetchone()
|
||
return row["cnt"] > 0 if row else False
|
||
finally:
|
||
conn.close()
|
||
except Exception:
|
||
return False
|
||
|
||
# ------------------------------------------------------------------
|
||
# 手动 tick（API 端点触发）
|
||
# ------------------------------------------------------------------
|
||
|
||
async def manual_tick(self) -> Dict[str, Any]:
|
||
"""手动触发一次 tick"""
|
||
logger.info("Manual tick triggered")
|
||
result = await self.tick()
|
||
result["manual"] = True
|
||
return result
|