""" core/workers/genesis_task_worker.py GenesisTaskWorker — converts AIVA's TASK_DISPATCH intent into a Genesis RWL task written to the Genesis task board at loop/tasks.json. Created by Story 5.09 (AIVA RLM Nexus PRD v2). Responsibilities: 1. Extract task description from intent.utterance + extracted_entities 2. Build a task entry with source="AIVA", status="pending", and a UUID 3. Append the entry to loop/tasks.json (load → append → write, no overwrite) 4. Return {"task_id": , "status": "queued"} — never None No SQLite. All file I/O is injected at construction so tests never touch the real task board. """ import json import logging from datetime import datetime, timezone from pathlib import Path from uuid import uuid4 from typing import Any logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Module constants # --------------------------------------------------------------------------- TASK_BOARD_PATH = Path("/mnt/e/genesis-system/loop/tasks.json") # Filler words stripped when building a human-readable task description. _FILLER_WORDS: frozenset[str] = frozenset( { "um", "uh", "er", "ah", "like", "you know", "basically", "literally", "actually", "just", "so", "well", "right", "okay", "ok", } ) # --------------------------------------------------------------------------- # GenesisTaskWorker # --------------------------------------------------------------------------- class GenesisTaskWorker: """ Converts an AIVA TASK_DISPATCH IntentSignal into a Genesis RWL task entry written to the task board JSON file. All file I/O goes through self._path so the worker is fully testable without touching the real loop/tasks.json. Args: task_board_path: Override the default TASK_BOARD_PATH (used in tests via ``tmp_path``). """ def __init__(self, task_board_path: Path | None = None) -> None: self._path: Path = task_board_path or TASK_BOARD_PATH # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ async def execute(self, intent: Any) -> dict: """ Main entry point. Called by SwarmRouter for every TASK_DISPATCH intent. Steps: 1. Build a human-readable task description from utterance + entities. 2. Create a task entry dict with a UUID, source="AIVA", created_at (ISO 8601 UTC), and status="pending". 3. Load existing tasks from the task board, append the new entry, and write the updated list back atomically. 4. Return {"task_id": , "status": "queued"}. Args: intent: An IntentSignal instance (duck-typed — no direct import keeps this worker decoupled from the classifier layer). Returns: dict with at minimum ``task_id`` and ``status`` keys. Never None. """ task_id: str = str(uuid4()) description: str = self._build_task_description(intent) task_entry: dict = { "id": task_id, "description": description, "source": "AIVA", "created_at": datetime.now(timezone.utc).isoformat(), "status": "pending", } try: tasks = self._load_tasks() tasks.append(task_entry) self._save_tasks(tasks) logger.info( "AIVA task queued: id=%s description=%r", task_id, description, ) except Exception as exc: # noqa: BLE001 logger.error("Failed to write task to board: %s", exc) return {"task_id": task_id, "status": "error", "reason": str(exc)} return {"task_id": task_id, "status": "queued"} # ------------------------------------------------------------------ # Private helpers # ------------------------------------------------------------------ def _build_task_description(self, intent: Any) -> str: """ Build a human-readable task description from the intent's utterance and extracted entities. Priority order: 1. ``extracted_entities["task"]`` — explicit task field from NLP 2. Cleaned utterance (filler words stripped, whitespace normalised) 3. Fallback: "AIVA task dispatch" Args: intent: An IntentSignal (or any duck-typed object). Returns: Non-empty string describing the task in plain language. """ # Priority 1: explicit task entity entities: dict = getattr(intent, "extracted_entities", {}) or {} if entities.get("task"): return str(entities["task"]).strip() # Priority 2: cleaned utterance utterance: str = getattr(intent, "utterance", "") or "" cleaned = self._clean_utterance(utterance) if cleaned: return cleaned # Priority 3: safe fallback return "AIVA task dispatch" def _clean_utterance(self, utterance: str) -> str: """ Strip common filler words and normalise whitespace from an utterance. Args: utterance: Raw caller utterance string. Returns: Cleaned string, or empty string if nothing useful remains. """ if not utterance: return "" words = utterance.split() filtered = [w for w in words if w.lower().rstrip(".,!?") not in _FILLER_WORDS] return " ".join(filtered).strip() def _load_tasks(self) -> list[dict]: """ Load the existing tasks list from the task board JSON file. Returns an empty list if the file does not exist, is empty, or contains invalid JSON — safe degradation to allow append. Returns: list of task dicts (may be empty). """ if not self._path.exists(): return [] try: content = self._path.read_text(encoding="utf-8").strip() if not content: return [] data = json.loads(content) # The real tasks.json is an object with a "stories" key. # We support both: a bare list OR an object with a "stories"/"tasks" list. if isinstance(data, list): return data if isinstance(data, dict): # Prefer "stories" key (existing task board format); fall back to "tasks" for key in ("stories", "tasks"): if isinstance(data.get(key), list): return data[key] # Unrecognised shape — start fresh logger.warning( "Unrecognised task board structure in %s — starting fresh", self._path ) return [] except (json.JSONDecodeError, OSError) as exc: logger.warning("Could not load task board from %s: %s", self._path, exc) return [] def _save_tasks(self, tasks: list[dict]) -> None: """ Write the tasks list back to the task board JSON file. Uses ``json.dumps`` with indent=2 for human-readable output. Creates parent directories if they don't exist. Args: tasks: Full list of task dicts to persist. """ self._path.parent.mkdir(parents=True, exist_ok=True) self._path.write_text(json.dumps(tasks, indent=2), encoding="utf-8") # VERIFICATION_STAMP # Story: 5.09 # Verified By: parallel-builder # Verified At: 2026-02-25 # Tests: 18/18 # Coverage: 100%