# -*- coding: utf-8 -*- """ TASK GENERATOR =============== Erzeugt autonom Tasks fuer jeden Agenten basierend auf: - Revenue-Zielen - Deadlines - Systemzustand - Externen Triggern - Selbsterkenntnis (epistemic state) """ from __future__ import annotations import datetime import hashlib import json from pathlib import Path from typing import Any, Dict, List, Optional from autonomous.decision_matrix import Priority, Task, DecisionMatrix from autonomous.epistemic_state import EpistemicState, EpistemicStatus from autonomous.agent_registry import AgentRegistry, AgentProfile from autonomous.revenue_tracker import RevenueTracker, RevenueType, RevenueStatus class TaskGenerator: """ Erzeugt autonom Tasks basierend auf Systemzustand. Kein Warten — erkennt was getan werden muss. """ def __init__(self, registry: AgentRegistry, revenue: RevenueTracker, state_dir: Optional[Path] = None): self.registry = registry self.revenue = revenue self.state_dir = state_dir or (Path(__file__).parent.parent / "autonomous" / "state" / "tasks") self.state_dir.mkdir(parents=True, exist_ok=True) self._trigger_log: List[dict] = [] def generate_for_agent(self, agent_id: int, matrix: DecisionMatrix, epistemic: EpistemicState) -> List[Task]: """Erzeuge alle passenden Tasks fuer einen Agenten.""" profile = self.registry.get(agent_id) if not profile: return [] tasks = [] # 1. Revenue-basierte Tasks tasks.extend(self._generate_revenue_tasks(profile, matrix)) # 2. Deadline-basierte Tasks tasks.extend(self._generate_deadline_tasks(profile, matrix)) # 3. Epistemic-basierte Tasks (Unsicherheiten aufloesen) tasks.extend(self._generate_epistemic_tasks(profile, matrix, epistemic)) # 4. Zyklische Maintenance-Tasks tasks.extend(self._generate_maintenance_tasks(profile, matrix)) # 5. Naechste definierte Aktionen aus Profile tasks.extend(self._generate_next_action_tasks(profile, matrix)) return tasks def _generate_revenue_tasks(self, profile: AgentProfile, matrix: DecisionMatrix) -> List[Task]: """Revenue-getriebene Tasks.""" tasks = [] agent_revenue = self.revenue.get_by_agent(profile.agent_id) # Fuer jede Revenue-Entry in IN_PROGRESS: Follow-up Task for entry in agent_revenue: if entry.status == RevenueStatus.IN_PROGRESS: tasks.append(matrix.create_task( title=f"Follow-up: {entry.title}", description=f"Follow-up fuer Revenue-Eintrag (EUR {entry.amount_eur:,.0f}, Prob: {entry.probability:.0%})", priority=Priority.HIGH if entry.amount_eur > 100_000 else Priority.MEDIUM, revenue_impact_eur=entry.weighted_value, deadline=entry.deadline, metadata={"revenue_entry_id": entry.entry_id, "type": "revenue_follow_up"}, )) elif entry.status == RevenueStatus.IDENTIFIED: tasks.append(matrix.create_task( title=f"Kontaktiere: {entry.title}", description=f"Erstkontakt fuer {entry.title} (EUR {entry.amount_eur:,.0f})", priority=Priority.MEDIUM, revenue_impact_eur=entry.weighted_value, deadline=entry.deadline, metadata={"revenue_entry_id": entry.entry_id, "type": "revenue_outreach"}, )) # Wenn Revenue unter Ziel: neue Quellen erschliessen current = sum(e.weighted_value for e in agent_revenue) if current < profile.revenue_target_eur * 0.3: tasks.append(matrix.create_task( title=f"Neue Revenue-Quellen fuer {profile.name}", description=f"Revenue-Track bei EUR {current:,.0f} / Ziel EUR {profile.revenue_target_eur:,.0f}. Identifiziere neue Quellen.", priority=Priority.HIGH, revenue_impact_eur=profile.revenue_target_eur * 0.1, metadata={"type": "revenue_exploration"}, )) return tasks def _generate_deadline_tasks(self, profile: AgentProfile, matrix: DecisionMatrix) -> List[Task]: """Deadline-getriebene Tasks.""" tasks = [] now = datetime.datetime.now(datetime.timezone.utc) for entry in self.revenue.entries.values(): if not entry.deadline: continue try: dl = datetime.datetime.fromisoformat(entry.deadline) if dl.tzinfo is None: dl = dl.replace(tzinfo=datetime.timezone.utc) days_left = (dl - now).total_seconds() / 86400 if entry.agent_id == profile.agent_id: if days_left <= 0: tasks.append(matrix.create_task( title=f"UEBERFAELLIG: {entry.title}", description=f"Deadline ueberschritten! {entry.title} — Deadline war {entry.deadline}", priority=Priority.CRITICAL, revenue_impact_eur=entry.amount_eur, deadline=entry.deadline, metadata={"type": "deadline_overdue", "revenue_entry": entry.entry_id}, )) elif days_left <= 7: tasks.append(matrix.create_task( title=f"Deadline in {days_left:.0f}d: {entry.title}", description=f"Deadline in {days_left:.0f} Tagen: {entry.title}", priority=Priority.HIGH, revenue_impact_eur=entry.amount_eur, deadline=entry.deadline, metadata={"type": "deadline_urgent"}, )) elif days_left <= 30: tasks.append(matrix.create_task( title=f"Vorbereitung: {entry.title}", description=f"Deadline in {days_left:.0f} Tagen — Vorbereitung starten", priority=Priority.MEDIUM, revenue_impact_eur=entry.amount_eur, deadline=entry.deadline, metadata={"type": "deadline_prep"}, )) except (ValueError, TypeError): pass return tasks def _generate_epistemic_tasks(self, profile: AgentProfile, matrix: DecisionMatrix, epistemic: EpistemicState) -> List[Task]: """Tasks zum Aufloesen von Unsicherheiten.""" tasks = [] uncertainties = epistemic.get_uncertainties() for ub in uncertainties: tasks.append(matrix.create_task( title=f"Validiere: {ub.content[:60]}", description=f"Unsichere Ueberzeugung (Confidence: {ub.confidence:.2f}): {ub.content}", priority=Priority.MEDIUM if ub.confidence > 0.3 else Priority.HIGH, metadata={"type": "epistemic_validation", "belief_id": ub.belief_id}, )) # Wenn kappa niedrig: generelle Validierung if epistemic.kappa_score < 3.34: tasks.append(matrix.create_task( title="Erhoehe Kappa-Score", description=f"Aktueller kappa: {epistemic.kappa_score:.2f} (Ziel: 3.34). Sammle mehr Evidenz.", priority=Priority.HIGH, metadata={"type": "kappa_improvement"}, )) return tasks def _generate_maintenance_tasks(self, profile: AgentProfile, matrix: DecisionMatrix) -> List[Task]: """Regelmaessige Wartungs-Tasks.""" tasks = [] # Wochenreport tasks.append(matrix.create_task( title=f"Wochenreport: {profile.name}", description=f"Erstelle woechentlichen Report fuer {profile.name}", priority=Priority.LOW, metadata={"type": "weekly_report", "cyclic": True}, )) # Audit-Check tasks.append(matrix.create_task( title=f"Audit-Check: {profile.name}", description=f"Pruefe Audit-Chain Integritaet fuer {profile.name}", priority=Priority.LOW, metadata={"type": "audit_check", "cyclic": True}, )) return tasks def _generate_next_action_tasks(self, profile: AgentProfile, matrix: DecisionMatrix) -> List[Task]: """Tasks aus den definierten 'next_actions'.""" tasks = [] for i, action in enumerate(profile.next_actions): if not action: continue tasks.append(matrix.create_task( title=f"Aktion {i+1}: {action[:50]}", description=action, priority=Priority.HIGH if i == 0 else Priority.MEDIUM, metadata={"type": "next_action", "action_index": i}, )) return tasks def log_trigger(self, trigger: str, source: str, details: Dict[str, Any] = None) -> None: self._trigger_log.append({ "trigger": trigger, "source": source, "details": details or {}, "timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(), }) # Persist path = self.state_dir / "triggers.jsonl" with path.open("a", encoding="utf-8") as f: f.write(json.dumps(self._trigger_log[-1], ensure_ascii=False) + "\n")