Function file_lock

ms_agent/utils/utils.py:746–776

(lock_dir: str, filename: str, timeout: float = 120.0)

@contextmanager
def file_lock(lock_dir: str, filename: str, timeout: float = 120.0):
    os.makedirs(lock_dir, exist_ok=True)
    lock_file_name = filename.replace(os.sep, '_') + '.lock'
    lock_file_path = os.path.join(lock_dir, lock_file_name)

    # Acquire lock with timeout
    start_time = time.time()

    while True:
        try:
            lock_fd = os.open(lock_file_path,
                              os.O_CREAT | os.O_EXCL | os.O_WRONLY)
            os.write(lock_fd, f'{os.getpid()}'.encode())
            break
        except FileExistsError:
            if time.time() - start_time >= timeout:
                raise TimeoutError(
                    f'Failed to acquire lock for {filename} after {timeout} seconds'
                )
            time.sleep(0.1)  # Wait 100ms before retry

    try:
        yield
    finally:
        # Release lock
        if lock_fd is not None:
            os.close(lock_fd)
        try:
            os.remove(lock_file_path)
        except OSError:
            pass


def render_markdown_todo(md_path: str,