""" Skill : DELEGATE Délègue une tâche à un agent spécialisé via MQTT et attend sa réponse. Log le résultat et notifie agent1 en cas d'erreur. Vérifie les plages horaires (work_hours) et le blackout avant délégation. Commande : DELEGATE: | """ import json import time import threading from datetime import datetime from pathlib import Path import paho.mqtt.client as mqtt SKILL_NAME = "delegate" TRIGGER = "DELEGATE:" CONFIG_FILE = Path("/opt/agent/config/config.json") REGISTRY_FILE = Path("/opt/agent/config/agents_registry.json") BLACKOUT_FILE = Path("/opt/agent/config/blackout_hours.json") TIMEOUT = 120 ERROR_KEYWORDS = ["erreur", "error", "timeout", "échec", "failed", "cannot", "permission denied", "command not found", "no such file", "connexion refusée"] def _load(): cfg = json.loads(CONFIG_FILE.read_text()) registry = json.loads(REGISTRY_FILE.read_text()) return cfg, registry def _is_error(result: str) -> bool: lower = result.lower() if "[erreur" in lower or "exit code" in lower: return True return any(kw in lower for kw in ERROR_KEYWORDS) def _check_availability(agent_name: str, registry: dict) -> tuple: """Retourne (disponible: bool, message: str).""" now = datetime.now() # Vérifier blackout_hours.json try: blackouts = json.loads(BLACKOUT_FILE.read_text()) for b in blackouts: if not b.get("enabled", True): continue start = datetime.strptime(b["start"], "%H:%M").replace( year=now.year, month=now.month, day=now.day) end = datetime.strptime(b["end"], "%H:%M").replace( year=now.year, month=now.month, day=now.day) if start <= now <= end: return False, "Blackout actif ({}) : {}-{}. Tâche non exécutée.".format( b.get("label", "maintenance"), b["start"], b["end"]) except Exception: pass # Vérifier work_hours de l'agent agent = registry.get(agent_name, {}) wh = agent.get("work_hours") if wh and wh.get("enabled", True): start_str = wh.get("start", "00:00") end_str = wh.get("end", "23:59") start = datetime.strptime(start_str, "%H:%M").replace( year=now.year, month=now.month, day=now.day) end = datetime.strptime(end_str, "%H:%M").replace( year=now.year, month=now.month, day=now.day) # Vérifier le jour days = wh.get("days") if days: day_map = ["mon","tue","wed","thu","fri","sat","sun"] if day_map[now.weekday()] not in days: return False, "Agent {} ne travaille pas ce jour. Jours actifs : {}.".format( agent_name, ", ".join(days)) if not (start <= now <= end): return False, "Agent {} hors plage horaire ({}-{}).".format( agent_name, start_str, end_str) return True, "" def _notify_error(host: str, port: int, agent: str, task: str, result: str): """Publie l'erreur sur le topic d'erreurs pour que agent1 notifie l'utilisateur.""" try: payload = json.dumps({ "agent" : agent, "task" : task[:200], "error" : result[:500], "source" : "delegate" }) pub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="delegate_err_pub") pub.connect(host, port) pub.publish("agents/errors", payload) pub.disconnect() except Exception: pass def execute(args: str) -> str: if "|" not in args: return "Erreur : format attendu → DELEGATE: | " agent_name, _, task = args.partition("|") agent_name, task = agent_name.strip(), task.strip() cfg, registry = _load() if agent_name not in registry: available = ", ".join(registry.keys()) return "Agent inconnu : «{}». Agents disponibles : {}".format(agent_name, available) # Vérifier disponibilité (plages horaires + blackout) available, reason = _check_availability(agent_name, registry) if not available: print("[DELEGATE] {} indisponible : {}".format(agent_name, reason)) return "[INDISPONIBLE] {} : {}".format(agent_name, reason) agent = registry[agent_name] inbox = agent["mqtt_inbox"] outbox = agent["mqtt_outbox"] host = cfg.get("mqtt_host", "localhost") port = int(cfg.get("mqtt_port", 1883)) response_received = threading.Event() response_container = [] def on_message(client, userdata, msg): response_container.append(msg.payload.decode(errors="replace")) response_received.set() sub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="agent1_delegate_sub") sub.on_message = on_message sub.connect(host, port) sub.subscribe(outbox) sub.loop_start() pub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="agent1_delegate_pub") pub.connect(host, port) pub.publish(inbox, task) pub.disconnect() print("[DELEGATE] Tâche envoyée à {} : {}".format(agent_name, task[:80])) start = time.time() received = response_received.wait(timeout=TIMEOUT) duration = time.time() - start sub.loop_stop() sub.disconnect() from skills.reporting import log_execution if received and response_container: result = response_container[0] status = "error" if _is_error(result) else "success" log_execution("delegate", agent_name, task, status, result, duration) if status == "error": _notify_error(host, port, agent_name, task, result) print("[DELEGATE] Erreur détectée dans la réponse de {}".format(agent_name)) return "[{}] {}".format(agent_name, result) # Timeout timeout_msg = "Timeout : {} n'a pas répondu dans les {}s.".format(agent_name, TIMEOUT) log_execution("delegate", agent_name, task, "timeout", timeout_msg, duration) _notify_error(host, port, agent_name, task, timeout_msg) return timeout_msg