Task Management — agent-core to Exo Mapping

# Agent-core: async-locked manager + executor registry
task_manager = TaskManager(config)
await task_manager.create_task("Research topic", priority=5)
scheduler = TaskScheduler(task_manager, executors={...})
await scheduler.run()

# Exo: Pydantic models + semaphore scheduler
from exo.task_controller import (
    TaskManager, TaskScheduler, TaskEventBus,
    IntentRecognizer, TaskLoopQueue,
)

bus = TaskEventBus()
manager = TaskManager(event_bus=bus, auto_complete_parent=True)
task = manager.create("Research topic", priority=5)
scheduler = TaskScheduler(manager, max_concurrent=3)
await scheduler.schedule(executor=my_agent_executor)

# Agent-core
task_manager = TaskManager(config)
parent = await task_manager.create_task("Analyze Q4 Results", priority=5)
child1 = await task_manager.create_task(
    "Gather data", priority=3, parent_id=parent.id,
)
child2 = await task_manager.create_task(
    "Run analysis", priority=4, parent_id=parent.id,
)
child3 = await task_manager.create_task(
    "Write summary", priority=2, parent_id=parent.id,
)

scheduler = TaskScheduler(task_manager, max_concurrent=2)
await scheduler.run()  # Picks up SUBMITTED tasks by priority

# Exo
from exo.task_controller import (
    TaskManager, TaskScheduler, TaskEventBus,
)

bus = TaskEventBus()
manager = TaskManager(event_bus=bus, auto_complete_parent=True)

parent = manager.create("Analyze Q4 Results", priority=5)
child1 = manager.create("Gather data", priority=3, parent_id=parent.id)
child2 = manager.create("Run analysis", priority=4, parent_id=parent.id)
child3 = manager.create("Write summary", priority=2, parent_id=parent.id)

scheduler = TaskScheduler(manager, max_concurrent=2)

async def my_executor(task):
    # Your agent logic here
    print(f"Executing: {task.name}")

await scheduler.schedule(executor=my_executor)
# Runs child2 (pri 4) and child1 (pri 3) first, then child3 (pri 2)
# When all children complete, parent auto-transitions to COMPLETED

# Agent-core
recognizer = IntentRecognizer(llm=my_llm)
intent = await recognizer.recognize("pause the data gathering task")
# intent.action == "pause_task", intent.task_id == child1.id

# Exo
from exo.task_controller import IntentRecognizer

recognizer = IntentRecognizer(model="openai:gpt-4o")
intent = await recognizer.recognize(
    "pause the data gathering task",
    available_tasks=[
        {"id": child1.id, "name": "Gather data"},
        {"id": child2.id, "name": "Run analysis"},
    ],
)
# intent.action == "pause_task"
# intent.task_id == child1.id
# intent.confidence == 0.95

# Agent-core
event_queue = EventQueue(session_id="session-1")
event_queue.subscribe("task.completed", on_task_done)
# Events emitted implicitly by task lifecycle

# Exo
from exo.task_controller import (
    TaskEventBus, TaskEventType, TaskEvent,
)

bus = TaskEventBus()

async def on_task_done(event: TaskEvent) -> None:
    print(f"Task {event.task_id} completed: {event.data}")

bus.subscribe(TaskEventType.COMPLETED, on_task_done)

# Events emitted automatically by TaskManager when event_bus is set
manager = TaskManager(event_bus=bus)
task = manager.create("My task")
manager.update(task.id, status=TaskStatus.WORKING)
manager.update(task.id, status=TaskStatus.COMPLETED)
# on_task_done is called with TaskEvent(event_type=COMPLETED, task_id=task.id)

from exo.task_controller import (
    TaskLoopQueue, TaskLoopEvent, TaskLoopEventType,
    get_task_loop_tools,
)

queue = TaskLoopQueue()

# Push a steering event from an external thread
queue.push(TaskLoopEvent(
    type=TaskLoopEventType.STEER,
    content="Focus on cost analysis instead",
))

# Agent loop checks queue each iteration
event = queue.pop()
if event and event.type == TaskLoopEventType.ABORT:
    raise SystemExit("Agent aborted")
elif event and event.type == TaskLoopEventType.STEER:
    # Redirect agent behavior
    pass

# Or use the built-in tools (bind queue at runtime)
tools = get_task_loop_tools()
for tool in tools:
    tool.bind(queue)