zelos.execution_trace

 1"""
 2Execution Trace — Complete execution timeline for a Goal.
 3
 4v0.9.0: Provides get_goal_trace() API returning a structured timeline
 5of every Task lifecycle event with optional input/output loading.
 6Supports pagination for large Goals.
 7"""
 8
 9from dataclasses import dataclass, field
10
11
12@dataclass
13class TraceEvent:
14    """A single event in a Task's execution timeline."""
15
16    event_type: str       # task.created / task.started / task.completed / task.failed / task.retry_scheduled
17    timestamp: float
18    payload: dict = field(default_factory=dict)
19
20    def to_dict(self) -> dict:
21        return {
22            "event_type": self.event_type,
23            "timestamp": self.timestamp,
24            "payload": dict(self.payload),
25        }
26
27
28@dataclass
29class TaskTrace:
30    """Complete execution trace for a single Task within a Goal."""
31
32    task_id: str
33    description: str = ""
34    required_capability: str = ""
35    agent_id: str | None = None
36    agent_name: str | None = None
37    status: str = "created"
38    attempt: int = 0
39    timeline: list[TraceEvent] = field(default_factory=list)
40    input_context: dict | None = None        # lazy-loaded
41    output_artifact: dict | None = None      # lazy-loaded
42    error: dict | None = None
43
44    def to_dict(self) -> dict:
45        return {
46            "task_id": self.task_id,
47            "description": self.description,
48            "required_capability": self.required_capability,
49            "agent_id": self.agent_id,
50            "agent_name": self.agent_name,
51            "status": self.status,
52            "attempt": self.attempt,
53            "timeline": [t.to_dict() for t in self.timeline],
54            "input_context": self.input_context,
55            "output_artifact": self.output_artifact,
56            "error": self.error,
57        }
58
59
60@dataclass
61class ExecutionTrace:
62    """Complete execution trace for a Goal — all Task traces merged.
63
64    Supports pagination via limit/offset.
65    """
66
67    goal_id: str
68    goal_description: str = ""
69    status: str = "unknown"
70    total_duration_ms: float = 0.0
71    tasks: list[TaskTrace] = field(default_factory=list)
72    total_tasks: int = 0    # Total before pagination
73
74    def to_dict(self) -> dict:
75        return {
76            "goal_id": self.goal_id,
77            "goal_description": self.goal_description,
78            "status": self.status,
79            "total_duration_ms": self.total_duration_ms,
80            "tasks": [t.to_dict() for t in self.tasks],
81            "total_tasks": self.total_tasks,
82        }
83
84    def paginate(self, limit: int = 50, offset: int = 0) -> "ExecutionTrace":
85        """Return a paginated view of this trace."""
86        sliced = ExecutionTrace(
87            goal_id=self.goal_id,
88            goal_description=self.goal_description,
89            status=self.status,
90            total_duration_ms=self.total_duration_ms,
91            tasks=self.tasks[offset:offset + limit],
92            total_tasks=len(self.tasks),
93        )
94        return sliced