51 lines
1.5 KiB
Python
51 lines
1.5 KiB
Python
"""Task Center orchestrator.
|
|
|
|
Coordinates task ingestion, persistence, lifecycle updates, events and statistics.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from .task_event_bridge import TaskEventBridge
|
|
from .task_source_ingestor import TaskSourceIngestor
|
|
from .task_statistics import TaskStatistics
|
|
from .task_store import TaskStore
|
|
|
|
|
|
class TaskOrchestrator:
|
|
def __init__(self):
|
|
self.ingestor = TaskSourceIngestor()
|
|
self.store = TaskStore()
|
|
self.events = TaskEventBridge()
|
|
self.statistics = TaskStatistics()
|
|
|
|
def create_from_command(self, path: str, bucket: str = "active") -> dict[str, Any]:
|
|
task = self.ingestor.command_file_to_task(path, bucket)
|
|
self.store.append(task)
|
|
self.events.task_created(task)
|
|
return task
|
|
|
|
def create_from_agent_execution(self, record: dict[str, Any]) -> dict[str, Any]:
|
|
task = self.ingestor.agent_execution_to_task(record)
|
|
self.store.append(task)
|
|
self.events.task_created(task)
|
|
return task
|
|
|
|
def sync_event(self, event: dict[str, Any]) -> dict[str, Any] | None:
|
|
task = self.ingestor.event_to_task(event)
|
|
if task:
|
|
self.store.append(task)
|
|
self.events.task_updated(task, event.get("kind", "event"))
|
|
return task
|
|
|
|
def dashboard(self) -> dict[str, Any]:
|
|
tasks = self.store.latest()
|
|
return {
|
|
"tasks": tasks,
|
|
"statistics": self.statistics.summarize(tasks),
|
|
}
|
|
|
|
|
|
__all__ = ["TaskOrchestrator"]
|