core/execution/executor.py

"""
BASYX V11 EXECUTOR
------------------

Central task execution engine.

Responsibilities:
- Load task from registry
- Create execution context
- Execute task safely
- Capture outputs
- Handle failures
- Support chaining
"""

from __future__ import annotations

import asyncio
import inspect
from typing import Dict, Any, List

from core.execution.context import create_context, ExecutionContext
from core.registry.loader import get_task_map


# =========================================================
# TASK CACHE
# =========================================================

TASK_MAP = get_task_map()


# =========================================================
# INTERNAL EXECUTION
# =========================================================

async def _run_task(
    task_name: str,
    inputs: Dict[str, Any],
) -> Dict[str, Any]:
    """
    Execute a single task safely.
    """

    if task_name not in TASK_MAP:
        raise ValueError(f"Unknown task: {task_name}")

    task = TASK_MAP[task_name]

    ctx: ExecutionContext = create_context(
        task_name=task_name,
        inputs=inputs,
    )

    ctx.mark_running()
    ctx.log("Starting task")

    try:

        # ---------------------------------------------
        # Execute task
        # ---------------------------------------------
        result = task.run

        if inspect.iscoroutinefunction(result):
            await result(ctx)
        else:
            await asyncio.to_thread(result, ctx)

        ctx.mark_complete()
        ctx.log("Task completed")

    except Exception as e:
        ctx.mark_failed(e)
        ctx.log(f"Task failed: {e}")

    return ctx.result()


# =========================================================
# PUBLIC EXECUTOR
# =========================================================

async def execute_task(
    task_name: str,
    inputs: Dict[str, Any],
) -> Dict[str, Any]:
    """
    Main entrypoint used by API + UI.
    """

    return await _run_task(task_name, inputs)


# =========================================================
# PIPELINE EXECUTION (CHAINED TASKS)
# =========================================================

async def execute_pipeline(
    tasks: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
    """
    Execute tasks sequentially.

    Example:
    [
        {"task": "transcribe", "inputs": {...}},
        {"task": "subtitles"},
        {"task": "render"}
    ]
    """

    results = []
    shared_memory = {}

    for step in tasks:

        name = step["task"]
        inputs = step.get("inputs", {})

        # Inject memory from previous step
        inputs["memory"] = shared_memory

        result = await _run_task(name, inputs)

        results.append(result)

        if result["status"] != "completed":
            break

        # propagate outputs
        shared_memory.update(result.get("outputs", {}))

    return results


# =========================================================
# PARALLEL EXECUTION
# =========================================================

async def execute_parallel(
    tasks: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
    """
    Run multiple tasks concurrently.
    """

    coroutines = [
        _run_task(t["task"], t.get("inputs", {}))
        for t in tasks
    ]

    return await asyncio.gather(*coroutines)


# =========================================================
# REGISTRY HOT RELOAD (DEV MODE)
# =========================================================

def reload_tasks():
    """
    Reload registry without restarting server.
    Useful during development.
    """
    global TASK_MAP
    TASK_MAP = get_task_map()