Process a parsed event and update workflow state.
(self, event: ParsedEvent)
| 201 | return None |
| 202 | |
| 203 | def _process_event(self, event: ParsedEvent): |
| 204 | """Process a parsed event and update workflow state.""" |
| 205 | data = event.data |
| 206 | etype = event.type |
| 207 | |
| 208 | if etype == "workflow_start": |
| 209 | self.workflow.task_name = data.get("task_name", "") |
| 210 | self.workflow.goal = data.get("goal", "") |
| 211 | self.workflow.model = data.get("model", "") |
| 212 | self.workflow.start_time = event.timestamp |
| 213 | self.workflow.status = "running" |
| 214 | self.workflow.total_tokens = 0 |
| 215 | self.workflow.prompt_tokens = 0 |
| 216 | self.workflow.completion_tokens = 0 |
| 217 | self.workflow.reasoning_tokens = 0 |
| 218 | self.workflow.cost = 0.0 |
| 219 | |
| 220 | elif etype == "workflow_end": |
| 221 | self.workflow.total_tokens = data.get("total_tokens", 0) |
| 222 | self.workflow.prompt_tokens = data.get("prompt_tokens", 0) |
| 223 | self.workflow.completion_tokens = data.get("completion_tokens", 0) |
| 224 | self.workflow.reasoning_tokens = data.get("reasoning_tokens", 0) |
| 225 | self.workflow.cost = float(data.get("cost", 0) or 0) |
| 226 | self.workflow.end_time = event.timestamp |
| 227 | self.workflow.status = "completed" |
| 228 | |
| 229 | elif etype == "step_start": |
| 230 | step_id = str(data.get("step_id") or data.get("step_number") or "?") |
| 231 | step = self._get_step(step_id) |
| 232 | step.number = str(data.get("step_number", step_id)) |
| 233 | step.name = data.get("name", step.name) |
| 234 | step.step_type = data.get("step_type", step.step_type) |
| 235 | step.instruction = data.get("instruction", step.instruction) |
| 236 | step.depth = int(data.get("depth", step.depth or 0)) |
| 237 | step.parent_step_id = data.get("parent_step_id") or step.parent_step_id |
| 238 | if data.get("parent_iteration") is not None: |
| 239 | try: |
| 240 | step.parent_iteration = int(data.get("parent_iteration")) |
| 241 | except Exception: |
| 242 | step.parent_iteration = step.parent_iteration |
| 243 | step.variables = data.get("variables", {}) |
| 244 | step.start_time = event.timestamp |
| 245 | step.status = "running" |
| 246 | self._current_step = step |
| 247 | |
| 248 | elif etype == "step_end": |
| 249 | step_id = str(data.get("step_id") or data.get("step_number") or "?") |
| 250 | step = self._step_index.get(step_id) |
| 251 | if step: |
| 252 | step.tokens = data.get("tokens", 0) |
| 253 | step.end_time = event.timestamp |
| 254 | step.status = data.get("status", "completed") |
| 255 | if self._current_step and self._current_step.step_id == step_id: |
| 256 | self._current_step = None |
| 257 | |
| 258 | elif etype == "agent_iteration": |
| 259 | step_id = str(data.get("step_id") or "?") |
| 260 | iteration_number = int(data.get("number", 0)) |
no test coverage detected