"""Seed the built-in functional agents used by strategy selection. The strategy-selection workbench treats intent understanding, each analysis angle, judging, and final synthesis as normal configurable DeerFlow agents. Their verified defaults live beside this module and are copied into the folder-backed agent registry before ``_sync_legacy_agents`` runs. Existing directories are preserved, except for an exact known legacy default whose content hash proves that an administrator has not customized it. """ from __future__ import annotations import logging import os from collections.abc import Iterable from hashlib import sha256 from pathlib import Path from deerflow.config.paths import get_paths logger = logging.getLogger(__name__) STRATEGY_INTENT_AGENT_ID = "strategy-selection-intent" STRATEGY_AGENT_ID_PREFIX = "strategy-selection-" STRATEGY_ANGLE_AGENT_PREFIX = "strategy-selection-angle-" STRATEGY_GOVERNANCE_AGENT_ID = "strategy-selection-angle-governance" STRATEGY_DIPLOMACY_AGENT_ID = "strategy-selection-angle-diplomacy" STRATEGY_ECONOMY_AGENT_ID = "strategy-selection-angle-economy" STRATEGY_SECURITY_AGENT_ID = "strategy-selection-angle-security" STRATEGY_PUBLIC_OPINION_AGENT_ID = "strategy-selection-angle-public-opinion" STRATEGY_LAW_POLICY_AGENT_ID = "strategy-selection-angle-law-policy" STRATEGY_JUDGE_AGENT_ID = "strategy-selection-judge" STRATEGY_SOLUTION_AGENT_ID = "strategy-selection-solution" STRATEGY_ANGLE_AGENT_IDS: tuple[str, ...] = ( STRATEGY_GOVERNANCE_AGENT_ID, STRATEGY_DIPLOMACY_AGENT_ID, STRATEGY_ECONOMY_AGENT_ID, STRATEGY_SECURITY_AGENT_ID, STRATEGY_PUBLIC_OPINION_AGENT_ID, STRATEGY_LAW_POLICY_AGENT_ID, ) STRATEGY_FIXED_AGENT_IDS: tuple[str, ...] = ( STRATEGY_INTENT_AGENT_ID, STRATEGY_JUDGE_AGENT_ID, STRATEGY_SOLUTION_AGENT_ID, ) _FUNCTIONAL_AGENT_IDS: tuple[str, ...] = ( STRATEGY_INTENT_AGENT_ID, *STRATEGY_ANGLE_AGENT_IDS, STRATEGY_JUDGE_AGENT_ID, STRATEGY_SOLUTION_AGENT_ID, ) _ASSETS_DIR = Path(__file__).parent / "_strategy_selection_seed_assets" _SEED_FILES: tuple[str, ...] = ("config.yaml", "SOUL.md") # The first shipped intent SOUL embedded the six built-in angle ids. Upgrade # only that byte-for-byte semantic default so existing installations adopt the # administrator-driven angle catalog without overwriting customized SOUL. _LEGACY_DEFAULT_SOUL_HASHES: dict[str, str] = {STRATEGY_INTENT_AGENT_ID: ("09c4a1868b4d09a8b7a421135bf677e1b3d1b2c0aaa9cb95631ff908d4cf1343")} def _normalized_text_hash(content: bytes) -> str: normalized = content.decode("utf-8").replace("\r\n", "\n").encode("utf-8") return sha256(normalized).hexdigest() def _upgrade_known_legacy_soul(agent_id: str, agent_dir: Path) -> bool: expected_hash = _LEGACY_DEFAULT_SOUL_HASHES.get(agent_id) target = agent_dir / "SOUL.md" source = _ASSETS_DIR / agent_id / "SOUL.md" if not expected_hash or not target.is_file() or not source.is_file(): return False temporary: Path | None = None try: if _normalized_text_hash(target.read_bytes()) != expected_hash: return False replacement = source.read_bytes() if _normalized_text_hash(replacement) == expected_hash: return False temporary = target.with_name(f".{target.name}.{os.getpid()}.seed-upgrade.tmp") temporary.write_bytes(replacement) os.replace(temporary, target) logger.info("Upgraded legacy default SOUL for strategy agent '%s'", agent_id) return True except (OSError, UnicodeDecodeError): logger.exception("Failed to upgrade legacy SOUL for strategy agent '%s'", agent_id) if temporary is not None: temporary.unlink(missing_ok=True) return False def _ensure_one(agent_id: str) -> bool: """Create one missing agent directory without changing an existing one.""" agent_dir = get_paths().agent_dir(agent_id) if agent_dir.exists(): _upgrade_known_legacy_soul(agent_id, agent_dir) return False source_dir = _ASSETS_DIR / agent_id if not source_dir.is_dir(): logger.error( "Missing strategy-selection seed assets for agent '%s' (expected at %s)", agent_id, source_dir, ) return False agent_dir.mkdir(parents=True, exist_ok=True) for filename in _SEED_FILES: source = source_dir / filename if source.exists(): (agent_dir / filename).write_bytes(source.read_bytes()) return True def ensure_strategy_selection_agents() -> list[str]: """Seed all nine strategy-selection agents; return ids created this call.""" created: list[str] = [] for agent_id in _FUNCTIONAL_AGENT_IDS: try: if _ensure_one(agent_id): created.append(agent_id) except Exception: logger.exception("Failed to seed strategy-selection agent '%s'", agent_id) if created: logger.info("Seeded missing strategy-selection agents: %s", ", ".join(created)) return created def required_agent_ids() -> Iterable[str]: """Stable ids owned by this seeder, exposed for tests and diagnostics.""" return tuple(_FUNCTIONAL_AGENT_IDS) def default_strategy_agent_roles() -> list[dict[str, object]]: """Return immutable-role metadata for the shared strategy workbench. The registry is deliberately separate from the folder-backed agent record: it owns role/category and lifecycle state, while model, SOUL and skills keep using the common agent configuration path. ``ensure_defaults`` only inserts missing rows, so an administrator disabling an angle is respected. """ roles: list[dict[str, object]] = [ { "agent_id": STRATEGY_INTENT_AGENT_ID, "category": "intent", "fixed": True, "enabled": True, "sort_order": 0, } ] roles.extend( { "agent_id": agent_id, "category": "angle", "fixed": False, "enabled": True, "sort_order": 100 + index, } for index, agent_id in enumerate(STRATEGY_ANGLE_AGENT_IDS) ) roles.extend( ( { "agent_id": STRATEGY_JUDGE_AGENT_ID, "category": "judge", "fixed": True, "enabled": True, "sort_order": 200, }, { "agent_id": STRATEGY_SOLUTION_AGENT_ID, "category": "solution", "fixed": True, "enabled": True, "sort_order": 300, }, ) ) return roles __all__ = [ "STRATEGY_AGENT_ID_PREFIX", "STRATEGY_ANGLE_AGENT_IDS", "STRATEGY_ANGLE_AGENT_PREFIX", "STRATEGY_DIPLOMACY_AGENT_ID", "STRATEGY_ECONOMY_AGENT_ID", "STRATEGY_FIXED_AGENT_IDS", "STRATEGY_GOVERNANCE_AGENT_ID", "STRATEGY_INTENT_AGENT_ID", "STRATEGY_JUDGE_AGENT_ID", "STRATEGY_LAW_POLICY_AGENT_ID", "STRATEGY_PUBLIC_OPINION_AGENT_ID", "STRATEGY_SECURITY_AGENT_ID", "STRATEGY_SOLUTION_AGENT_ID", "default_strategy_agent_roles", "ensure_strategy_selection_agents", "required_agent_ids", ]