Function workspace_locks

backend/app/services/workspace_locking.py:57–80

(
    agent_id: uuid.UUID,
    paths: list[str],
    *,
    ttl_seconds: int = DEFAULT_LOCK_TTL_SECONDS,
)

@asynccontextmanager
async def workspace_locks(
    agent_id: uuid.UUID,
    paths: list[str],
    *,
    ttl_seconds: int = DEFAULT_LOCK_TTL_SECONDS,
):
    normalized = sorted({_normalize_workspace_path(path) or "." for path in paths if path is not None})
    owner_token = uuid.uuid4().hex
    acquired: list[str] = []
    try:
        for path in normalized:
            ok = await acquire_workspace_lock(
                agent_id,
                path,
                owner_token=owner_token,
                ttl_seconds=ttl_seconds,
            )
            if not ok:
                raise RuntimeError(f"Workspace lock busy: {path}")
            acquired.append(path)
        yield
    finally:
        for path in reversed(acquired):
            await release_workspace_lock(agent_id, path, owner_token=owner_token)