File size: 3,640 Bytes
345855e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 | """
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() |