"""Seed the built-in task-analysis coordinator agent (任务分析总控). ``task-analysis-coordinator`` 是「任务分析定时执行」的专用总控单例:定时触发后 先由它理解任务、给编排中的每个席位分解具体任务(无人值守,禁止澄清),再由 ChainBindingExecutor 按阶段执行席位。业务语义与多智能体会商的 ``roundtable-coordinator``(有人在线、允许澄清)不同,因此独立内置,互不影响。 模式与 ``_roundtable_seed`` 一致:验证版 config.yaml + SOUL.md 作为源码树资源 (``_task_analysis_seed_assets/``,随 git 发布),seeder 字节级复制到 ``.deer-flow/agents//``,目录存在不覆盖(尊重现场调优)。落 DB 由启动期的 ``_sync_legacy_agents`` 完成。用途标签由 ``_builtin_agent_tag_seed`` 负责。 """ from __future__ import annotations import logging from pathlib import Path from deerflow.config.paths import get_paths logger = logging.getLogger(__name__) TASK_ANALYSIS_COORDINATOR_ID = "task-analysis-coordinator" _ASSETS_DIR = Path(__file__).parent / "_task_analysis_seed_assets" _SEED_FILES: tuple[str, ...] = ("config.yaml", "SOUL.md") def ensure_task_analysis_functional_agents() -> list[str]: """Seed the built-in task-analysis coordinator when missing. Idempotent. Called from the startup lifespan and defensively from the executor before each coordinator round, so a fresh deployment (``.deer-flow/`` is gitignored) or an admin-deleted directory self-heals. """ created: list[str] = [] src_dir = _ASSETS_DIR / TASK_ANALYSIS_COORDINATOR_ID if not src_dir.is_dir(): logger.error( "Missing seed assets for task-analysis agent '%s' (expected at %s)", TASK_ANALYSIS_COORDINATOR_ID, src_dir, ) return created agent_dir = get_paths().agent_dir(TASK_ANALYSIS_COORDINATOR_ID) if agent_dir.exists(): return created try: agent_dir.mkdir(parents=True, exist_ok=True) for fname in _SEED_FILES: src = src_dir / fname if src.exists(): (agent_dir / fname).write_bytes(src.read_bytes()) created.append(TASK_ANALYSIS_COORDINATOR_ID) logger.info("Seeded task-analysis coordinator agent '%s'", TASK_ANALYSIS_COORDINATOR_ID) except Exception: logger.exception( "Failed to seed task-analysis agent '%s'", TASK_ANALYSIS_COORDINATOR_ID ) return created