Clean up orphaned lock files from previous server runs. Scans all registered projects for .agent.lock files and removes them if the referenced process is no longer running. Returns: Number of orphaned lock files cleaned up
()
| 562 | |
| 563 | |
| 564 | def cleanup_orphaned_locks() -> int: |
| 565 | """ |
| 566 | Clean up orphaned lock files from previous server runs. |
| 567 | |
| 568 | Scans all registered projects for .agent.lock files and removes them |
| 569 | if the referenced process is no longer running. |
| 570 | |
| 571 | Returns: |
| 572 | Number of orphaned lock files cleaned up |
| 573 | """ |
| 574 | import sys |
| 575 | root = Path(__file__).parent.parent.parent |
| 576 | if str(root) not in sys.path: |
| 577 | sys.path.insert(0, str(root)) |
| 578 | |
| 579 | from registry import list_registered_projects |
| 580 | |
| 581 | cleaned = 0 |
| 582 | try: |
| 583 | projects = list_registered_projects() |
| 584 | for name, info in projects.items(): |
| 585 | project_path = Path(info.get("path", "")) |
| 586 | if not project_path.exists(): |
| 587 | continue |
| 588 | |
| 589 | # Check both legacy and new locations for lock files |
| 590 | from autoforge_paths import get_autoforge_dir |
| 591 | lock_locations = [ |
| 592 | project_path / ".agent.lock", |
| 593 | get_autoforge_dir(project_path) / ".agent.lock", |
| 594 | ] |
| 595 | lock_file = None |
| 596 | for candidate in lock_locations: |
| 597 | if candidate.exists(): |
| 598 | lock_file = candidate |
| 599 | break |
| 600 | if lock_file is None: |
| 601 | continue |
| 602 | |
| 603 | try: |
| 604 | lock_content = lock_file.read_text().strip() |
| 605 | # Support both legacy format (just PID) and new format (PID:CREATE_TIME) |
| 606 | if ":" in lock_content: |
| 607 | pid_str, create_time_str = lock_content.split(":", 1) |
| 608 | pid = int(pid_str) |
| 609 | stored_create_time = float(create_time_str) |
| 610 | else: |
| 611 | # Legacy format - just PID |
| 612 | pid = int(lock_content) |
| 613 | stored_create_time = None |
| 614 | |
| 615 | # Check if process is still running |
| 616 | if psutil.pid_exists(pid): |
| 617 | try: |
| 618 | proc = psutil.Process(pid) |
| 619 | # Verify it's the same process using creation time (handles PID reuse) |
| 620 | if stored_create_time is not None: |
| 621 | if abs(proc.create_time() - stored_create_time) > 1.0: |
no test coverage detected