Check if the agent process is still alive. Updates status to 'crashed' if process has died unexpectedly. Returns: True if healthy, False otherwise
(self)
| 487 | return False, f"Failed to resume agent: {e}" |
| 488 | |
| 489 | async def healthcheck(self) -> bool: |
| 490 | """ |
| 491 | Check if the agent process is still alive. |
| 492 | |
| 493 | Updates status to 'crashed' if process has died unexpectedly. |
| 494 | |
| 495 | Returns: |
| 496 | True if healthy, False otherwise |
| 497 | """ |
| 498 | if not self.process: |
| 499 | return self.status == "stopped" |
| 500 | |
| 501 | poll = self.process.poll() |
| 502 | if poll is not None: |
| 503 | # Process has terminated |
| 504 | if self.status in ("running", "paused"): |
| 505 | self.status = "crashed" |
| 506 | self._remove_lock() |
| 507 | return False |
| 508 | |
| 509 | return True |
| 510 | |
| 511 | def get_status_dict(self) -> dict: |
| 512 | """Get current status as a dictionary.""" |
no test coverage detected