Function list_locks

ci/util/lock_admin.py:30–93

List all locks in cache directory with status.

(cache_dir: Path)

def list_locks(cache_dir: Path) -> int:
    """List all locks in cache directory with status."""
    locks = list_active_locks(cache_dir)

    # Also show all locks from the database
    db = get_lock_database()
    all_db_locks = db.list_all_locks()

    if not locks and not all_db_locks:
        print("No locks found")
        return 0

    print(f"\n📋 ACTIVE LOCKS")
    print("=" * 80)

    active_count = 0
    stale_count = 0

    # Show locks from cache_dir filtering
    for lock in locks:
        status_icon = "🔴 STALE" if lock.is_stale else "🟢 ACTIVE"
        pid = lock.pid if lock.pid else "unknown"
        operation = lock.operation
        timestamp = lock.timestamp if lock.timestamp else "unknown"
        hostname = lock.hostname if lock.hostname else "unknown"

        if lock.is_stale:
            stale_count += 1
        else:
            active_count += 1

        print(f"\n{status_icon}")
        print(f"  Lock: {lock.path}")
        print(f"  PID: {pid}")
        print(f"  Operation: {operation}")
        print(f"  Timestamp: {timestamp}")
        print(f"  Hostname: {hostname}")

    # Show any DB locks not already shown
    shown_names = {lock.path for lock in locks}
    for db_lock in all_db_locks:
        lock_name = db_lock["lock_name"]
        if lock_name not in shown_names:
            pid = db_lock["owner_pid"]
            alive = is_process_alive(pid)
            status_icon = "🟢 ACTIVE" if alive else "🔴 STALE"

            if alive:
                active_count += 1
            else:
                stale_count += 1

            print(f"\n{status_icon}")
            print(f"  Lock: {lock_name}")
            print(f"  PID: {pid}")
            print(f"  Operation: {db_lock['operation']}")
            print(f"  Mode: {db_lock['lock_mode']}")
            print(f"  Hostname: {db_lock['hostname']}")