Function file_lock

src/codex_self_evolution/storage.py:272–289

(paths: Paths, name: str = "compile.lock", stale_after_seconds: int = DEFAULT_LOCK_STALE_SECONDS)

@contextmanager
def file_lock(paths: Paths, name: str = "compile.lock", stale_after_seconds: int = DEFAULT_LOCK_STALE_SECONDS) -> Iterator[Path]:
    ensure_runtime_dirs(paths)
    lock_path = compiler_lock_path(paths, name=name)
    status = lock_status(paths, stale_after_seconds=stale_after_seconds, name=name)
    if status["locked"] and not status["stale"]:
        raise CompileLockError(f"compile already locked: {lock_path}")
    if status["locked"] and status["stale"] and lock_path.exists():
        lock_path.unlink()
    payload = {
        "created_at": utc_now().replace(microsecond=0).isoformat().replace("+00:00", "Z"),
        "pid": os.getpid(),
    }
    atomic_write_json(lock_path, payload)
    try:
        yield lock_path
    finally:
        if lock_path.exists():
            lock_path.unlink()