160 lines
5.6 KiB
Python
160 lines
5.6 KiB
Python
"""Task source ingestion for AIOS Task Center.
|
|
|
|
Converts Command Watcher files, OpenClaw/Hermes/Codex execution records and
|
|
Generic Agent Event Bus records into normalized Task Center task dictionaries.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict, dataclass, field
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
def utc_now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TaskSourceRecord:
|
|
source_type: str
|
|
source_id: str
|
|
title: str
|
|
assigned_by: str
|
|
executor: str
|
|
status: str
|
|
priority: str = "normal"
|
|
created_at: str = field(default_factory=utc_now)
|
|
updated_at: str = field(default_factory=utc_now)
|
|
evidence: list[str] = field(default_factory=list)
|
|
metadata: dict[str, Any] = field(default_factory=dict)
|
|
|
|
def to_task(self) -> dict[str, Any]:
|
|
return {
|
|
"task_id": f"task-{self.source_type}-{self.source_id}",
|
|
"title": self.title,
|
|
"assigned_by": self.assigned_by,
|
|
"executor": self.executor,
|
|
"source": self.source_type,
|
|
"status": self.status,
|
|
"priority": self.priority,
|
|
"created_at": self.created_at,
|
|
"updated_at": self.updated_at,
|
|
"progress": {
|
|
"percent": self._progress_percent(),
|
|
"current_step": self.status,
|
|
"completed_steps": [],
|
|
"remaining_steps": [],
|
|
"blocked_reason": self.metadata.get("blocked_reason"),
|
|
},
|
|
"notification": {
|
|
"required": True,
|
|
"sent": bool(self.metadata.get("notification_sent")),
|
|
"channel": self.metadata.get("notification_channel"),
|
|
"sent_at": self.metadata.get("notification_sent_at"),
|
|
},
|
|
"evidence": self.evidence,
|
|
"metadata": self.metadata,
|
|
}
|
|
|
|
def _progress_percent(self) -> int:
|
|
mapping = {
|
|
"new": 0,
|
|
"pending": 5,
|
|
"assigned": 15,
|
|
"running": 50,
|
|
"blocked": 50,
|
|
"waiting_approval": 60,
|
|
"approved": 70,
|
|
"completed": 100,
|
|
"closed": 100,
|
|
"failed": 100,
|
|
"cancelled": 100,
|
|
}
|
|
return mapping.get(self.status, 0)
|
|
|
|
|
|
class TaskSourceIngestor:
|
|
COMMAND_STATUS_MAP = {
|
|
"active": "pending",
|
|
"running": "running",
|
|
"completed": "completed",
|
|
"failed": "failed",
|
|
"archive": "closed",
|
|
}
|
|
|
|
def command_file_to_task(self, path: str, bucket: str = "active") -> dict[str, Any]:
|
|
p = Path(path)
|
|
title = p.stem
|
|
return TaskSourceRecord(
|
|
source_type="command_watcher",
|
|
source_id=title,
|
|
title=title,
|
|
assigned_by="command-watcher",
|
|
executor=self._infer_executor(title),
|
|
status=self.COMMAND_STATUS_MAP.get(bucket, "pending"),
|
|
priority="high" if "P0" in title or "fix" in title.lower() else "normal",
|
|
evidence=[str(p)],
|
|
metadata={"bucket": bucket},
|
|
).to_task()
|
|
|
|
def agent_execution_to_task(self, record: dict[str, Any]) -> dict[str, Any]:
|
|
source_id = record.get("execution_id", record.get("id", "unknown"))
|
|
status = record.get("status", "running")
|
|
return TaskSourceRecord(
|
|
source_type="agent_execution",
|
|
source_id=source_id,
|
|
title=record.get("title", f"Agent execution {source_id}"),
|
|
assigned_by=record.get("assigned_by", record.get("entry_agent", "agent-system")),
|
|
executor=record.get("closed_by", record.get("entry_agent", "unknown-agent")),
|
|
status=status,
|
|
priority=record.get("priority", "normal"),
|
|
created_at=record.get("started_at", utc_now()),
|
|
updated_at=record.get("ended_at", utc_now()),
|
|
evidence=[record.get("execution_id", source_id)],
|
|
metadata=record,
|
|
).to_task()
|
|
|
|
def event_to_task(self, event: dict[str, Any]) -> dict[str, Any] | None:
|
|
kind = event.get("kind")
|
|
payload = event.get("payload", {})
|
|
if kind not in {"task_created", "task_updated", "agent_started", "agent_completed", "agent_failed", "search_started", "search_completed"}:
|
|
return None
|
|
source_id = event.get("correlation_id") or event.get("id", "event")
|
|
status = self._status_from_event(kind, payload)
|
|
return TaskSourceRecord(
|
|
source_type="event_bus",
|
|
source_id=source_id,
|
|
title=payload.get("title") or event.get("summary", kind),
|
|
assigned_by=event.get("source", "event-bus"),
|
|
executor=event.get("target", "task-center"),
|
|
status=status,
|
|
created_at=event.get("created_at", utc_now()),
|
|
updated_at=event.get("created_at", utc_now()),
|
|
evidence=[event.get("id", source_id)],
|
|
metadata={"event_kind": kind, "payload": payload},
|
|
).to_task()
|
|
|
|
def _status_from_event(self, kind: str, payload: dict[str, Any]) -> str:
|
|
if kind.endswith("completed"):
|
|
return "completed"
|
|
if kind.endswith("failed"):
|
|
return "failed"
|
|
return payload.get("status", "running")
|
|
|
|
def _infer_executor(self, title: str) -> str:
|
|
lower = title.lower()
|
|
if "openclaw" in lower:
|
|
return "openclaw"
|
|
if "hermes" in lower:
|
|
return "hermes"
|
|
if "codex" in lower:
|
|
return "codex"
|
|
if "search" in lower:
|
|
return "openclaw"
|
|
return "openclaw"
|
|
|
|
|
|
__all__ = ["TaskSourceRecord", "TaskSourceIngestor"]
|